| """ |
| 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 os |
| import shutil |
| import tempfile |
| import unittest |
| from pathlib import Path |
| |
| import pyarrow as pa |
| |
| from pypaimon import CatalogFactory |
| from pypaimon.common.file_io import FileIO |
| from pypaimon.read.reader.format_blob_reader import FormatBlobReader |
| from pypaimon.schema.data_types import AtomicType, DataField |
| from pypaimon.table.row.blob import Blob, BlobData, BlobRef, BlobDescriptor |
| from pypaimon.table.row.generic_row import GenericRowDeserializer, GenericRowSerializer, GenericRow |
| from pypaimon.table.row.row_kind import RowKind |
| |
| |
| class MockFileIO: |
| """Mock FileIO for testing.""" |
| |
| def __init__(self, file_io: FileIO): |
| self._file_io = file_io |
| |
| def get_file_size(self, path: str) -> int: |
| """Get file size.""" |
| return self._file_io.get_file_size(Path(path)) |
| |
| def new_input_stream(self, path: Path): |
| """Create new input stream for reading.""" |
| return self._file_io.new_input_stream(path) |
| |
| |
| class BlobTest(unittest.TestCase): |
| """Tests for Blob interface following org.apache.paimon.data.BlobTest.""" |
| |
| def setUp(self): |
| """Set up test environment with temporary file.""" |
| # Create a temporary directory and file |
| self.temp_dir = tempfile.mkdtemp() |
| self.file = os.path.join(self.temp_dir, "test.txt") |
| |
| # Write test data to the file |
| with open(self.file, 'wb') as f: |
| f.write(b"test data") |
| |
| def tearDown(self): |
| """Clean up temporary files.""" |
| try: |
| if os.path.exists(self.file): |
| os.remove(self.file) |
| os.rmdir(self.temp_dir) |
| except OSError: |
| pass # Ignore cleanup errors |
| |
| def test_from_data(self): |
| """Test Blob.from_data() method.""" |
| test_data = b"test data" |
| blob = Blob.from_data(test_data) |
| |
| # Verify it returns a BlobData instance |
| self.assertIsInstance(blob, BlobData) |
| |
| # Verify the data matches |
| self.assertEqual(blob.to_data(), test_data) |
| |
| def test_from_local(self): |
| """Test Blob.from_local() method.""" |
| blob = Blob.from_local(self.file) |
| |
| # Verify it returns a BlobRef instance |
| self.assertIsInstance(blob, BlobRef) |
| |
| # Verify the data matches |
| self.assertEqual(blob.to_data(), b"test data") |
| |
| def test_from_file_with_offset_and_length(self): |
| """Test Blob.from_file() method with offset and length.""" |
| file_io = FileIO(self.file if self.file.startswith('file://') else f"file://{self.file}", {}) |
| blob = Blob.from_file(file_io, self.file, 0, 4) |
| |
| # Verify it returns a BlobRef instance |
| self.assertIsInstance(blob, BlobRef) |
| |
| # Verify the data matches (first 4 bytes: "test") |
| self.assertEqual(blob.to_data(), b"test") |
| |
| def test_from_http(self): |
| """Test Blob.from_http() method.""" |
| uri = "http://example.com/file.txt" |
| blob = Blob.from_http(uri) |
| |
| # Verify it returns a BlobRef instance |
| self.assertIsInstance(blob, BlobRef) |
| |
| # Verify the descriptor has the correct URI |
| descriptor = blob.to_descriptor() |
| self.assertEqual(descriptor.uri, uri) |
| self.assertEqual(descriptor.offset, 0) |
| self.assertEqual(descriptor.length, -1) |
| |
| def test_blob_data_interface_compliance(self): |
| """Test that BlobData properly implements Blob interface.""" |
| test_data = b"interface test data" |
| blob_data = BlobData(test_data) |
| |
| # Test that it's a Blob |
| self.assertIsInstance(blob_data, Blob) |
| |
| # Test interface methods |
| self.assertEqual(blob_data.to_data(), test_data) |
| |
| # Test to_descriptor raises RuntimeError |
| with self.assertRaises(RuntimeError) as context: |
| blob_data.to_descriptor() |
| self.assertIn("Blob data can not convert to descriptor", str(context.exception)) |
| |
| # Test new_input_stream |
| stream = blob_data.new_input_stream() |
| self.assertEqual(stream.read(), test_data) |
| stream.close() |
| |
| def test_blob_ref_interface_compliance(self): |
| """Test that BlobRef properly implements Blob interface.""" |
| blob_ref = Blob.from_local(self.file) |
| |
| # Test that it's a Blob |
| self.assertIsInstance(blob_ref, Blob) |
| |
| # Test interface methods |
| self.assertEqual(blob_ref.to_data(), b"test data") |
| |
| # Test to_descriptor returns valid descriptor |
| descriptor = blob_ref.to_descriptor() |
| self.assertEqual(descriptor.uri, self.file) |
| self.assertEqual(descriptor.offset, 0) |
| self.assertEqual(descriptor.length, -1) |
| |
| # Test new_input_stream |
| stream = blob_ref.new_input_stream() |
| self.assertEqual(stream.read(), b"test data") |
| stream.close() |
| |
| def test_blob_equality_and_hashing(self): |
| """Test blob equality and hashing behavior.""" |
| # Test BlobData equality |
| data1 = BlobData(b"same data") |
| data2 = BlobData(b"same data") |
| data3 = BlobData(b"different data") |
| |
| self.assertEqual(data1, data2) |
| self.assertNotEqual(data1, data3) |
| self.assertEqual(hash(data1), hash(data2)) |
| |
| # Test BlobRef equality |
| ref1 = Blob.from_local(self.file) |
| ref2 = Blob.from_local(self.file) |
| |
| self.assertEqual(ref1, ref2) |
| self.assertEqual(hash(ref1), hash(ref2)) |
| |
| def test_blob_factory_methods_return_correct_types(self): |
| """Test that all factory methods return the expected types.""" |
| # from_data should return BlobData |
| blob_data = Blob.from_data(b"test") |
| self.assertIsInstance(blob_data, BlobData) |
| self.assertIsInstance(blob_data, Blob) |
| |
| # from_local should return BlobRef |
| blob_ref = Blob.from_local(self.file) |
| self.assertIsInstance(blob_ref, BlobRef) |
| self.assertIsInstance(blob_ref, Blob) |
| |
| # from_file should return BlobRef |
| file_io = FileIO(self.file if self.file.startswith('file://') else f"file://{self.file}", {}) |
| blob_file = Blob.from_file(file_io, self.file, 0, os.path.getsize(self.file)) |
| self.assertIsInstance(blob_file, BlobRef) |
| self.assertIsInstance(blob_file, Blob) |
| |
| # from_http should return BlobRef |
| blob_http = Blob.from_http("http://example.com/test.bin") |
| self.assertIsInstance(blob_http, BlobRef) |
| self.assertIsInstance(blob_http, Blob) |
| |
| def test_blob_data_convenience_methods(self): |
| # Test from_bytes class method |
| blob2 = BlobData.from_bytes(b"from bytes") |
| self.assertEqual(blob2.to_data(), b"from bytes") |
| |
| def test_generic_row_deserializer_parse_blob(self): |
| """Test GenericRowDeserializer._parse_blob method.""" |
| # Create test data with BLOB field |
| test_blob_data = b"Test BLOB data for parsing" |
| blob_data = BlobData(test_blob_data) |
| |
| # Create fields with BLOB type |
| fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "blob_field", AtomicType("BLOB")), |
| ] |
| |
| # Create and serialize a row with blob data |
| original_row = GenericRow([42, blob_data], fields, RowKind.INSERT) |
| serialized_bytes = GenericRowSerializer.to_bytes(original_row) |
| |
| # Test the full deserialization process (which uses _parse_blob internally) |
| deserialized_row = GenericRowDeserializer.from_bytes(serialized_bytes, fields) |
| |
| # Verify the deserialized blob |
| deserialized_blob = deserialized_row.values[1] |
| self.assertIsInstance(deserialized_blob, BlobData) |
| self.assertEqual(deserialized_blob.to_data(), test_blob_data) |
| |
| # Test with empty blob data |
| empty_blob = BlobData(b"") |
| empty_row = GenericRow([1, empty_blob], fields, RowKind.INSERT) |
| empty_serialized = GenericRowSerializer.to_bytes(empty_row) |
| empty_deserialized = GenericRowDeserializer.from_bytes(empty_serialized, fields) |
| |
| empty_deserialized_blob = empty_deserialized.values[1] |
| self.assertIsInstance(empty_deserialized_blob, BlobData) |
| self.assertEqual(empty_deserialized_blob.to_data(), b"") |
| |
| # Test with binary data containing null bytes |
| binary_blob_data = b"\x00\x01\x02\x03\xff\xfe\xfd" |
| binary_blob = BlobData(binary_blob_data) |
| binary_row = GenericRow([99, binary_blob], fields, RowKind.INSERT) |
| binary_serialized = GenericRowSerializer.to_bytes(binary_row) |
| binary_deserialized = GenericRowDeserializer.from_bytes(binary_serialized, fields) |
| |
| binary_deserialized_blob = binary_deserialized.values[1] |
| self.assertIsInstance(binary_deserialized_blob, BlobData) |
| self.assertEqual(binary_deserialized_blob.to_data(), binary_blob_data) |
| |
| def test_generic_row_deserializer_parse_blob_with_multiple_fields(self): |
| """Test _parse_blob with multiple BLOB fields in a row.""" |
| # Create test data with multiple BLOB fields |
| blob1_data = b"First BLOB data" |
| blob2_data = b"Second BLOB with different content" |
| blob3_data = b"" # Empty blob |
| |
| blob1 = BlobData(blob1_data) |
| blob2 = BlobData(blob2_data) |
| blob3 = BlobData(blob3_data) |
| |
| # Create fields with multiple BLOB types |
| fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "name", AtomicType("STRING")), |
| DataField(2, "blob1", AtomicType("BLOB")), |
| DataField(3, "blob2", AtomicType("BLOB")), |
| DataField(4, "blob3", AtomicType("BLOB")), |
| ] |
| |
| # Create and serialize a row with multiple blobs |
| original_row = GenericRow([123, "test_row", blob1, blob2, blob3], fields, RowKind.INSERT) |
| serialized_bytes = GenericRowSerializer.to_bytes(original_row) |
| |
| # Deserialize and verify all blobs |
| deserialized_row = GenericRowDeserializer.from_bytes(serialized_bytes, fields) |
| |
| # Verify each blob field |
| self.assertEqual(deserialized_row.values[0], 123) |
| self.assertEqual(deserialized_row.values[1], "test_row") |
| |
| deserialized_blob1 = deserialized_row.values[2] |
| self.assertIsInstance(deserialized_blob1, BlobData) |
| self.assertEqual(deserialized_blob1.to_data(), blob1_data) |
| |
| deserialized_blob2 = deserialized_row.values[3] |
| self.assertIsInstance(deserialized_blob2, BlobData) |
| self.assertEqual(deserialized_blob2.to_data(), blob2_data) |
| |
| deserialized_blob3 = deserialized_row.values[4] |
| self.assertIsInstance(deserialized_blob3, BlobData) |
| self.assertEqual(deserialized_blob3.to_data(), blob3_data) |
| |
| def test_generic_row_deserializer_parse_blob_with_null_values(self): |
| """Test _parse_blob with null BLOB values.""" |
| # Create fields with BLOB type |
| fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "blob_field", AtomicType("BLOB")), |
| DataField(2, "name", AtomicType("STRING")), |
| ] |
| |
| # Create row with null blob (None value) |
| original_row = GenericRow([456, None, "test_with_null"], fields, RowKind.INSERT) |
| serialized_bytes = GenericRowSerializer.to_bytes(original_row) |
| |
| # Deserialize and verify null blob is handled correctly |
| deserialized_row = GenericRowDeserializer.from_bytes(serialized_bytes, fields) |
| |
| self.assertEqual(deserialized_row.values[0], 456) |
| self.assertIsNone(deserialized_row.values[1]) # Null blob should remain None |
| self.assertEqual(deserialized_row.values[2], "test_with_null") |
| |
| def test_generic_row_deserializer_parse_blob_large_data(self): |
| """Test _parse_blob with large BLOB data.""" |
| # Create large blob data (1MB) |
| large_blob_data = b"X" * (1024 * 1024) # 1MB of 'X' characters |
| large_blob = BlobData(large_blob_data) |
| |
| fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "large_blob", AtomicType("BLOB")), |
| ] |
| |
| # Create and serialize row with large blob |
| original_row = GenericRow([789, large_blob], fields, RowKind.INSERT) |
| serialized_bytes = GenericRowSerializer.to_bytes(original_row) |
| |
| # Deserialize and verify large blob |
| deserialized_row = GenericRowDeserializer.from_bytes(serialized_bytes, fields) |
| |
| deserialized_large_blob = deserialized_row.values[1] |
| self.assertIsInstance(deserialized_large_blob, BlobData) |
| self.assertEqual(len(deserialized_large_blob.to_data()), 1024 * 1024) |
| self.assertEqual(deserialized_large_blob.to_data(), large_blob_data) |
| |
| def test_blob_descriptor_creation(self): |
| """Test BlobDescriptor creation and properties.""" |
| # Test basic creation |
| descriptor = BlobDescriptor("test://example.uri", 100, 200) |
| |
| self.assertEqual(descriptor.uri, "test://example.uri") |
| self.assertEqual(descriptor.offset, 100) |
| self.assertEqual(descriptor.length, 200) |
| self.assertEqual(descriptor.version, BlobDescriptor.CURRENT_VERSION) |
| |
| def test_blob_descriptor_creation_with_version(self): |
| """Test BlobDescriptor creation with explicit version.""" |
| descriptor = BlobDescriptor("test://example.uri", 50, 150, version=2) |
| |
| self.assertEqual(descriptor.uri, "test://example.uri") |
| self.assertEqual(descriptor.offset, 50) |
| self.assertEqual(descriptor.length, 150) |
| self.assertEqual(descriptor.version, 2) |
| |
| def test_blob_descriptor_serialization_deserialization(self): |
| """Test BlobDescriptor serialization and deserialization.""" |
| # Test with various URIs and parameters |
| test_cases = [ |
| ("file:///path/to/file.bin", 0, -1), |
| ("https://example.com/data.blob", 1024, 2048), |
| ("s3://bucket/key", 0, 1000000), |
| ("test://simple", 42, 84), |
| ] |
| |
| for uri, offset, length in test_cases: |
| with self.subTest(uri=uri, offset=offset, length=length): |
| # Create original descriptor |
| original = BlobDescriptor(uri, offset, length) |
| |
| # Serialize |
| serialized = original.serialize() |
| self.assertIsInstance(serialized, bytes) |
| self.assertGreater(len(serialized), 0) |
| |
| # Deserialize |
| deserialized = BlobDescriptor.deserialize(serialized) |
| |
| # Verify equality |
| self.assertEqual(deserialized, original) |
| self.assertEqual(deserialized.uri, uri) |
| self.assertEqual(deserialized.offset, offset) |
| self.assertEqual(deserialized.length, length) |
| self.assertEqual(deserialized.version, BlobDescriptor.CURRENT_VERSION) |
| |
| def test_blob_descriptor_serialization_with_unicode(self): |
| """Test BlobDescriptor serialization with Unicode characters.""" |
| # Test with Unicode characters in URI |
| unicode_uri = "file:///测试/文件.bin" |
| descriptor = BlobDescriptor(unicode_uri, 0, 100) |
| |
| # Serialize and deserialize |
| serialized = descriptor.serialize() |
| deserialized = BlobDescriptor.deserialize(serialized) |
| |
| # Verify Unicode is preserved |
| self.assertEqual(deserialized.uri, unicode_uri) |
| self.assertEqual(deserialized, descriptor) |
| |
| def test_blob_descriptor_deserialization_invalid_data(self): |
| """Test BlobDescriptor deserialization with invalid data.""" |
| # Test with too short data |
| with self.assertRaises(ValueError) as context: |
| BlobDescriptor.deserialize(b"sho") # Only 3 bytes, need at least 5 |
| self.assertIn("too short", str(context.exception)) |
| |
| # Test with invalid version (version 0) |
| # Create valid data but with wrong version |
| valid_descriptor = BlobDescriptor("test://uri", 0, 100) |
| valid_data = bytearray(valid_descriptor.serialize()) |
| valid_data[0] = 0 # Set invalid version (0) |
| |
| with self.assertRaises(ValueError) as context: |
| BlobDescriptor.deserialize(bytes(valid_data)) |
| self.assertIn("Unsupported BlobDescriptor version", str(context.exception)) |
| |
| # Test with incomplete data (missing URI bytes) |
| incomplete_data = b'\x01\x00\x00\x00\x10' # Version 1, URI length 16, but no URI bytes |
| with self.assertRaises(ValueError) as context: |
| BlobDescriptor.deserialize(incomplete_data) |
| self.assertIn("URI length exceeds data size", str(context.exception)) |
| |
| def test_blob_descriptor_equality_and_hashing(self): |
| """Test BlobDescriptor equality and hashing.""" |
| # Create identical descriptors |
| desc1 = BlobDescriptor("test://uri", 100, 200) |
| desc2 = BlobDescriptor("test://uri", 100, 200) |
| desc3 = BlobDescriptor("test://uri", 100, 201) # Different length |
| desc4 = BlobDescriptor("test://other", 100, 200) # Different URI |
| |
| # Test equality |
| self.assertEqual(desc1, desc2) |
| self.assertNotEqual(desc1, desc3) |
| self.assertNotEqual(desc1, desc4) |
| self.assertNotEqual(desc1, None) |
| self.assertNotEqual(desc1, "not a descriptor") |
| |
| # Test hashing |
| self.assertEqual(hash(desc1), hash(desc2)) |
| # Hash should be different for different descriptors (though not guaranteed) |
| self.assertNotEqual(hash(desc1), hash(desc3)) |
| self.assertNotEqual(hash(desc1), hash(desc4)) |
| |
| def test_blob_descriptor_string_representation(self): |
| """Test BlobDescriptor string representation.""" |
| descriptor = BlobDescriptor("test://example.uri", 42, 84) |
| |
| str_repr = str(descriptor) |
| self.assertIn("test://example.uri", str_repr) |
| self.assertIn("42", str_repr) |
| self.assertIn("84", str_repr) |
| self.assertIn("BlobDescriptor", str_repr) |
| |
| # __repr__ should be the same as __str__ |
| self.assertEqual(str_repr, repr(descriptor)) |
| |
| def test_blob_descriptor_version_handling(self): |
| """Test BlobDescriptor version handling.""" |
| # Test current version |
| descriptor = BlobDescriptor("test://uri", 0, 100) |
| self.assertEqual(descriptor.version, BlobDescriptor.CURRENT_VERSION) |
| |
| # Test explicit version |
| descriptor_v2 = BlobDescriptor("test://uri", 0, 100, version=2) |
| self.assertEqual(descriptor_v2.version, 2) |
| |
| # Serialize and deserialize should preserve version |
| serialized = descriptor_v2.serialize() |
| deserialized = BlobDescriptor.deserialize(serialized) |
| self.assertEqual(deserialized.version, 2) |
| |
| def test_blob_descriptor_edge_cases(self): |
| """Test BlobDescriptor with edge cases.""" |
| # Test with empty URI |
| empty_uri_desc = BlobDescriptor("", 0, 0) |
| serialized = empty_uri_desc.serialize() |
| deserialized = BlobDescriptor.deserialize(serialized) |
| self.assertEqual(deserialized.uri, "") |
| |
| # Test with very long URI |
| long_uri = "file://" + "a" * 1000 + "/file.bin" |
| long_uri_desc = BlobDescriptor(long_uri, 0, 1000000) |
| serialized = long_uri_desc.serialize() |
| deserialized = BlobDescriptor.deserialize(serialized) |
| self.assertEqual(deserialized.uri, long_uri) |
| |
| # Test with negative values |
| negative_desc = BlobDescriptor("test://uri", -1, -1) |
| serialized = negative_desc.serialize() |
| deserialized = BlobDescriptor.deserialize(serialized) |
| self.assertEqual(deserialized.offset, -1) |
| self.assertEqual(deserialized.length, -1) |
| |
| def test_blob_descriptor_with_blob_ref(self): |
| """Test BlobDescriptor integration with BlobRef.""" |
| # Create a descriptor |
| descriptor = BlobDescriptor(self.file, 0, -1) |
| |
| # Create BlobRef from descriptor |
| blob_ref = Blob.from_local(self.file) |
| |
| # Verify descriptor is preserved |
| returned_descriptor = blob_ref.to_descriptor() |
| self.assertEqual(returned_descriptor, descriptor) |
| |
| # Verify data can be read through BlobRef |
| data = blob_ref.to_data() |
| self.assertEqual(data, b"test data") |
| |
| def test_blob_descriptor_serialization_format(self): |
| """Test BlobDescriptor serialization format details.""" |
| descriptor = BlobDescriptor("test", 12345, 67890) |
| serialized = descriptor.serialize() |
| |
| # Check that serialized data starts with version byte |
| self.assertEqual(serialized[0], BlobDescriptor.CURRENT_VERSION) |
| |
| # Check minimum length (version + uri_length + uri + offset + length) |
| # 1 + 4 + len("test") + 8 + 8 = 25 bytes |
| self.assertEqual(len(serialized), 25) |
| |
| # Verify round-trip consistency |
| deserialized = BlobDescriptor.deserialize(serialized) |
| re_serialized = deserialized.serialize() |
| self.assertEqual(serialized, re_serialized) |
| |
| |
| class BlobEndToEndTest(unittest.TestCase): |
| """End-to-end tests for blob functionality with schema definition, file writing, and reading.""" |
| |
| def setUp(self): |
| """Set up test environment.""" |
| self.temp_dir = tempfile.mkdtemp() |
| self.warehouse = os.path.join(self.temp_dir, 'warehouse') |
| # Create catalog for table operations |
| self.catalog = CatalogFactory.create({ |
| 'warehouse': self.warehouse |
| }) |
| self.catalog.create_database('test_db', False) |
| |
| def tearDown(self): |
| """Clean up test environment.""" |
| try: |
| shutil.rmtree(self.temp_dir) |
| except OSError: |
| pass |
| |
| def test_blob_end_to_end(self): |
| # Set up file I/O |
| file_io = FileIO(self.temp_dir, {}) |
| |
| blob_field_name = "blob_field" |
| # ========== Step 1: Check Type Validation ========== |
| blob_fields = [DataField(0, blob_field_name, AtomicType("BLOB"))] |
| for blob_field in blob_fields: |
| self.assertIsInstance(blob_field.type, AtomicType) |
| self.assertEqual(blob_field.type.type, "BLOB") |
| |
| # ========== Step 2: Write Data ========== |
| test_data = {blob_field_name: BlobData(b'End-to-end test: PDF header %PDF-1.4\n...')} |
| blob_files = {} |
| blob_data = [test_data[blob_field_name].to_data()] |
| schema = pa.schema([pa.field(blob_field_name, pa.large_binary())]) |
| table = pa.table([blob_data], schema=schema) |
| blob_files[blob_field_name] = Path(self.temp_dir) / (blob_field_name + ".blob") |
| file_io.write_blob(blob_files[blob_field_name], table, False) |
| self.assertTrue(file_io.exists(blob_files[blob_field_name])) |
| |
| # ========== Step 3: Read Data and Check Data ========== |
| for field_name, file_path in blob_files.items(): |
| read_fields = blob_fields |
| reader = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(file_path), |
| read_fields=[field_name], |
| full_fields=read_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| |
| # Read data |
| batch = reader.read_arrow_batch() |
| self.assertIsNotNone(batch, f"{field_name} batch should not be None") |
| self.assertEqual(batch.num_rows, 1, f"{field_name} should have 1 row") |
| |
| # Verify data integrity |
| read_blob_data = batch.column(0)[0].as_py() |
| expected_blob_data = test_data[field_name].to_data() |
| self.assertEqual(read_blob_data, expected_blob_data, f"{field_name} data should match") |
| |
| reader.close() |
| |
| def test_blob_complex_types_throw_exception(self): |
| """Test that complex types containing BLOB elements throw exceptions during read/write operations.""" |
| from pypaimon.schema.data_types import DataField, AtomicType, ArrayType, MultisetType, MapType |
| from pypaimon.table.row.blob import BlobData |
| from pypaimon.common.file_io import FileIO |
| from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer |
| from pypaimon.table.row.row_kind import RowKind |
| from pathlib import Path |
| |
| # Set up file I/O |
| file_io = FileIO(self.temp_dir, {}) |
| |
| # ========== Test ArrayType(nullable=True, element_type=AtomicType("BLOB")) ========== |
| array_fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "blob_array", ArrayType(nullable=True, element_type=AtomicType("BLOB"))), |
| ] |
| |
| # Test serialization throws exception for ArrayType<BLOB> |
| array_blob_data = [ |
| BlobData(b"Array blob 1"), |
| BlobData(b"Array blob 2"), |
| BlobData(b"Array blob 3") |
| ] |
| |
| array_row = GenericRow([1, array_blob_data], array_fields, RowKind.INSERT) |
| |
| # GenericRowSerializer should throw exception for complex types |
| with self.assertRaises(ValueError) as context: |
| GenericRowSerializer.to_bytes(array_row) |
| self.assertIn("AtomicType", str(context.exception)) |
| |
| # Note: FileIO.write_blob validation for complex types is tested separately below |
| |
| # ========== Test MultisetType(nullable=True, element_type=AtomicType("BLOB")) ========== |
| multiset_fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "blob_multiset", MultisetType(nullable=True, element_type=AtomicType("BLOB"))), |
| ] |
| |
| # Test serialization throws exception for MultisetType<BLOB> |
| multiset_blob_data = [ |
| BlobData(b"Multiset blob 1"), |
| BlobData(b"Multiset blob 2"), |
| BlobData(b"Multiset blob 1"), # Duplicate allowed in multiset |
| ] |
| |
| multiset_row = GenericRow([2, multiset_blob_data], multiset_fields, RowKind.INSERT) |
| |
| # GenericRowSerializer should throw exception for complex types |
| with self.assertRaises(ValueError) as context: |
| GenericRowSerializer.to_bytes(multiset_row) |
| self.assertIn("AtomicType", str(context.exception)) |
| map_fields = [ |
| DataField(0, "id", AtomicType("INT")), |
| DataField(1, "blob_map", MapType( |
| nullable=True, key_type=AtomicType("STRING"), value_type=AtomicType("BLOB") |
| )), |
| ] |
| |
| # Test serialization throws exception for MapType<STRING, BLOB> |
| map_blob_data = { |
| "document": BlobData(b"Document content"), |
| "image": BlobData(b"Image data"), |
| "metadata": BlobData(b"Metadata content") |
| } |
| |
| map_row = GenericRow([3, map_blob_data], map_fields, RowKind.INSERT) |
| |
| # GenericRowSerializer should throw exception for complex types |
| with self.assertRaises(ValueError) as context: |
| GenericRowSerializer.to_bytes(map_row) |
| self.assertIn("AtomicType", str(context.exception)) |
| |
| # ========== Test FileIO.write_blob validation for complex types ========== |
| # Test that FileIO.write_blob properly validates and rejects complex types |
| |
| # Create a table with multiple columns (should fail - blob format requires single column) |
| multi_column_schema = pa.schema([ |
| pa.field("blob1", pa.large_binary()), |
| pa.field("blob2", pa.large_binary()) |
| ]) |
| multi_column_table = pa.table([ |
| [b"blob1_data"], |
| [b"blob2_data"] |
| ], schema=multi_column_schema) |
| |
| multi_column_file = Path(self.temp_dir) / "multi_column.blob" |
| |
| # Should throw RuntimeError for multiple columns |
| with self.assertRaises(RuntimeError) as context: |
| file_io.write_blob(multi_column_file, multi_column_table, False) |
| self.assertIn("single column", str(context.exception)) |
| |
| # Test that FileIO.write_blob rejects null values |
| null_schema = pa.schema([pa.field("blob_with_nulls", pa.large_binary())]) |
| null_table = pa.table([[b"data", None]], schema=null_schema) |
| |
| null_file = Path(self.temp_dir) / "null_data.blob" |
| |
| # Should throw RuntimeError for null values |
| with self.assertRaises(RuntimeError) as context: |
| file_io.write_blob(null_file, null_table, False) |
| self.assertIn("null values", str(context.exception)) |
| |
| # ========== Test FormatBlobReader with complex type schema ========== |
| # Create a valid blob file first |
| valid_blob_data = [b"Valid blob content"] |
| valid_schema = pa.schema([pa.field("valid_blob", pa.large_binary())]) |
| valid_table = pa.table([valid_blob_data], schema=valid_schema) |
| |
| valid_blob_file = Path(self.temp_dir) / "valid_blob.blob" |
| file_io.write_blob(valid_blob_file, valid_table, False) |
| |
| # Try to read with complex type field definition - this should fail |
| # because FormatBlobReader tries to create PyArrow schema with complex types |
| complex_read_fields = [ |
| DataField(0, "valid_blob", ArrayType(nullable=True, element_type=AtomicType("BLOB"))) |
| ] |
| |
| # FormatBlobReader creation should work, but reading should fail due to schema mismatch |
| reader = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(valid_blob_file), |
| read_fields=["valid_blob"], |
| full_fields=complex_read_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| |
| # Reading should fail because the schema expects complex type but data is atomic |
| with self.assertRaises(Exception) as context: |
| reader.read_arrow_batch() |
| # The error could be ArrowTypeError or other PyArrow-related errors |
| self.assertTrue( |
| "ArrowTypeError" in str(type(context.exception)) or |
| "TypeError" in str(type(context.exception)) or |
| "ValueError" in str(type(context.exception)) |
| ) |
| |
| reader.close() |
| |
| def test_blob_advanced_scenarios(self): |
| """Test advanced blob scenarios: corruption, truncation, zero-length, large blobs, compression, cross-format.""" |
| from pypaimon.schema.data_types import DataField, AtomicType |
| from pypaimon.common.file_io import FileIO |
| from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor |
| from pathlib import Path |
| |
| # Set up file I/O |
| file_io = FileIO(self.temp_dir, {}) |
| |
| # ========== Test 1: Corrupted file header test ========== |
| |
| # Create a valid blob file first |
| valid_blob_data = [b"Test blob content for corruption test"] |
| valid_schema = pa.schema([pa.field("test_blob", pa.large_binary())]) |
| valid_table = pa.table([valid_blob_data], schema=valid_schema) |
| |
| header_test_file = Path(self.temp_dir) / "header_test.blob" |
| file_io.write_blob(header_test_file, valid_table, False) |
| |
| # Read the file and corrupt the header (last 5 bytes: index_length + version) |
| with open(header_test_file, 'rb') as f: |
| original_data = f.read() |
| |
| # Corrupt the version byte (last byte) |
| corrupted_data = bytearray(original_data) |
| corrupted_data[-1] = 99 # Invalid version (should be 1) |
| |
| corrupted_header_file = Path(self.temp_dir) / "corrupted_header.blob" |
| with open(corrupted_header_file, 'wb') as f: |
| f.write(corrupted_data) |
| |
| # Try to read corrupted file - should detect invalid version |
| fields = [DataField(0, "test_blob", AtomicType("BLOB"))] |
| |
| # Reading should fail due to invalid version |
| with self.assertRaises(IOError) as context: |
| FormatBlobReader( |
| file_io=file_io, |
| file_path=str(corrupted_header_file), |
| read_fields=["test_blob"], |
| full_fields=fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| self.assertIn("Unsupported blob file version", str(context.exception)) |
| |
| # ========== Test 2: Truncated blob file (mid-blob) read ========== |
| |
| # Create a blob file with substantial content |
| large_content = b"Large blob content: " + b"X" * 1000 + b" End of content" |
| large_blob_data = [large_content] |
| large_schema = pa.schema([pa.field("large_blob", pa.large_binary())]) |
| large_table = pa.table([large_blob_data], schema=large_schema) |
| |
| full_blob_file = Path(self.temp_dir) / "full_blob.blob" |
| file_io.write_blob(full_blob_file, large_table, False) |
| |
| # Read the full file and truncate it in the middle |
| with open(full_blob_file, 'rb') as f: |
| full_data = f.read() |
| |
| # Truncate to about 50% of original size (mid-blob) |
| truncated_size = len(full_data) // 2 |
| truncated_data = full_data[:truncated_size] |
| |
| truncated_file = Path(self.temp_dir) / "truncated.blob" |
| with open(truncated_file, 'wb') as f: |
| f.write(truncated_data) |
| |
| # Try to read truncated file - should fail gracefully |
| with self.assertRaises((IOError, OSError)) as context: |
| FormatBlobReader( |
| file_io=file_io, |
| file_path=str(truncated_file), |
| read_fields=["large_blob"], |
| full_fields=fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| # Should detect truncation/incomplete data (either invalid header or invalid version) |
| self.assertTrue( |
| "cannot read header" in str(context.exception) or |
| "Unsupported blob file version" in str(context.exception) |
| ) |
| |
| # ========== Test 3: Zero-length blob handling ========== |
| |
| # Create blob with zero-length content |
| zero_blob_data = [b""] # Empty blob |
| zero_schema = pa.schema([pa.field("zero_blob", pa.large_binary())]) |
| zero_table = pa.table([zero_blob_data], schema=zero_schema) |
| |
| zero_blob_file = Path(self.temp_dir) / "zero_length.blob" |
| file_io.write_blob(zero_blob_file, zero_table, False) |
| |
| # Verify file was created |
| self.assertTrue(file_io.exists(zero_blob_file)) |
| file_size = file_io.get_file_size(zero_blob_file) |
| self.assertGreater(file_size, 0) # File should have headers even with empty blob |
| |
| # Read zero-length blob |
| zero_fields = [DataField(0, "zero_blob", AtomicType("BLOB"))] |
| zero_reader = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(zero_blob_file), |
| read_fields=["zero_blob"], |
| full_fields=zero_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| |
| zero_batch = zero_reader.read_arrow_batch() |
| self.assertIsNotNone(zero_batch) |
| self.assertEqual(zero_batch.num_rows, 1) |
| |
| # Verify empty blob content |
| read_zero_blob = zero_batch.column(0)[0].as_py() |
| self.assertEqual(read_zero_blob, b"") |
| self.assertEqual(len(read_zero_blob), 0) |
| zero_reader.close() |
| |
| # ========== Test 4: Large blob (multi-GB range) simulation ========== |
| # Simulate large blob without actually creating multi-GB data |
| # Test chunked writing and memory-safe reading patterns |
| |
| # Create moderately large blob (10MB) to test chunking behavior |
| chunk_size = 1024 * 1024 # 1MB chunks |
| large_blob_content = b"LARGE_BLOB_CHUNK:" + b"L" * (chunk_size - 17) # Fill to 1MB |
| |
| # Simulate multiple chunks |
| simulated_large_data = [large_blob_content * 10] # 10MB total |
| large_sim_schema = pa.schema([pa.field("large_sim_blob", pa.large_binary())]) |
| large_sim_table = pa.table([simulated_large_data], schema=large_sim_schema) |
| |
| large_sim_file = Path(self.temp_dir) / "large_simulation.blob" |
| file_io.write_blob(large_sim_file, large_sim_table, False) |
| |
| # Verify large file was written |
| large_sim_size = file_io.get_file_size(large_sim_file) |
| self.assertGreater(large_sim_size, 10 * 1024 * 1024) # Should be > 10MB |
| |
| # Read large blob in memory-safe manner |
| large_sim_fields = [DataField(0, "large_sim_blob", AtomicType("BLOB"))] |
| large_sim_reader = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(large_sim_file), |
| read_fields=["large_sim_blob"], |
| full_fields=large_sim_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| |
| large_sim_batch = large_sim_reader.read_arrow_batch() |
| self.assertIsNotNone(large_sim_batch) |
| self.assertEqual(large_sim_batch.num_rows, 1) |
| |
| # Verify large blob content (check prefix to avoid loading all into memory for comparison) |
| read_large_blob = large_sim_batch.column(0)[0].as_py() |
| self.assertTrue(read_large_blob.startswith(b"LARGE_BLOB_CHUNK:")) |
| self.assertEqual(len(read_large_blob), len(large_blob_content) * 10) |
| large_sim_reader.close() |
| |
| # ========== Test 5: Index compression/decompression validation ========== |
| # Test DeltaVarintCompressor roundtrip |
| test_indices = [0, 100, 250, 1000, 5000, 10000, 50000] |
| |
| # Compress indices |
| compressed_indices = DeltaVarintCompressor.compress(test_indices) |
| self.assertIsInstance(compressed_indices, bytes) |
| self.assertGreater(len(compressed_indices), 0) |
| |
| # Decompress indices |
| decompressed_indices = DeltaVarintCompressor.decompress(compressed_indices) |
| self.assertEqual(decompressed_indices, test_indices) |
| |
| # Test corruption detection in compressed indices |
| if len(compressed_indices) > 1: |
| # Corrupt the compressed data |
| corrupted_indices = bytearray(compressed_indices) |
| corrupted_indices[-1] = (corrupted_indices[-1] + 1) % 256 # Flip last byte |
| |
| # Decompression should fail or produce different results |
| try: |
| corrupted_result = DeltaVarintCompressor.decompress(bytes(corrupted_indices)) |
| # If decompression succeeds, result should be different |
| self.assertNotEqual(corrupted_result, test_indices) |
| except Exception: |
| pass |
| |
| # ========== Test 6: Cross-format guard (multi-field tables) ========== |
| # Test that blob format rejects multi-field tables |
| multi_field_schema = pa.schema([ |
| pa.field("blob_field", pa.large_binary()), |
| pa.field("string_field", pa.string()), |
| pa.field("int_field", pa.int64()) |
| ]) |
| |
| multi_field_table = pa.table([ |
| [b"blob_data_1", b"blob_data_2"], |
| ["string_1", "string_2"], |
| [100, 200] |
| ], schema=multi_field_schema) |
| |
| multi_field_file = Path(self.temp_dir) / "multi_field.blob" |
| |
| # Should reject multi-field table |
| with self.assertRaises(RuntimeError) as context: |
| file_io.write_blob(multi_field_file, multi_field_table, False) |
| self.assertIn("single column", str(context.exception)) |
| |
| # Test that blob format rejects non-binary field types |
| non_binary_schema = pa.schema([pa.field("string_field", pa.string())]) |
| non_binary_table = pa.table([["not_binary_data"]], schema=non_binary_schema) |
| |
| non_binary_file = Path(self.temp_dir) / "non_binary.blob" |
| |
| # Should reject non-binary field |
| with self.assertRaises(RuntimeError) as context: |
| file_io.write_blob(non_binary_file, non_binary_table, False) |
| # Should fail due to type conversion issues (non-binary field can't be converted to BLOB) |
| self.assertTrue( |
| "large_binary" in str(context.exception) or |
| "to_paimon_type" in str(context.exception) or |
| "missing" in str(context.exception) or |
| "Field must be Blob/BlobData instance" in str(context.exception) |
| ) |
| |
| # Test that blob format rejects tables with null values |
| null_schema = pa.schema([pa.field("blob_with_null", pa.large_binary())]) |
| null_table = pa.table([[b"data", None, b"more_data"]], schema=null_schema) |
| |
| null_file = Path(self.temp_dir) / "with_nulls.blob" |
| |
| # Should reject null values |
| with self.assertRaises(RuntimeError) as context: |
| file_io.write_blob(null_file, null_table, False) |
| self.assertIn("null values", str(context.exception)) |
| |
| def test_blob_end_to_end_with_descriptor(self): |
| # Set up file I/O |
| file_io = FileIO(self.temp_dir, {}) |
| |
| # ========== Step 1: Write data to local file ========== |
| # Create test data and write it to a local file |
| test_content = b'This is test blob content stored in an external file for descriptor testing.' |
| # Write the test content to a local file |
| local_data_file = Path(self.temp_dir) / "external_blob" |
| with open(local_data_file, 'wb') as f: |
| f.write(test_content) |
| # Verify the file was created and has the correct content |
| self.assertTrue(local_data_file.exists()) |
| with open(local_data_file, 'rb') as f: |
| written_content = f.read() |
| self.assertEqual(written_content, test_content) |
| |
| # ========== Step 2: Use this file as blob descriptor ========== |
| # Create a BlobDescriptor pointing to the local file |
| blob_descriptor = BlobDescriptor( |
| uri=str(local_data_file), |
| offset=0, |
| length=len(test_content) |
| ) |
| # Serialize the descriptor to bytes (this is what would be stored in the blob column) |
| descriptor_bytes = blob_descriptor.serialize() |
| self.assertIsInstance(descriptor_bytes, bytes) |
| self.assertGreater(len(descriptor_bytes), 0) |
| |
| # Create PyArrow table with the serialized descriptor |
| blob_field_name = "blob_descriptor_field" |
| schema = pa.schema([pa.field(blob_field_name, pa.large_binary())]) |
| table = pa.table([[descriptor_bytes]], schema=schema) |
| |
| # Write the blob file with blob_as_descriptor=True |
| blob_file_path = Path(self.temp_dir) / "descriptor_blob.blob" |
| file_io.write_blob(blob_file_path, table, blob_as_descriptor=True) |
| # Verify the blob file was created |
| self.assertTrue(file_io.exists(blob_file_path)) |
| file_size = file_io.get_file_size(blob_file_path) |
| self.assertGreater(file_size, 0) |
| |
| # ========== Step 3: Read data and check ========== |
| # Define schema for reading |
| read_fields = [DataField(0, blob_field_name, AtomicType("BLOB"))] |
| reader = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(blob_file_path), |
| read_fields=[blob_field_name], |
| full_fields=read_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=True |
| ) |
| |
| # Read the data with blob_as_descriptor=True (should return a descriptor) |
| batch = reader.read_arrow_batch() |
| self.assertIsNotNone(batch) |
| self.assertEqual(batch.num_rows, 1) |
| self.assertEqual(batch.num_columns, 1) |
| |
| read_blob_bytes = batch.column(0)[0].as_py() |
| self.assertIsInstance(read_blob_bytes, bytes) |
| |
| # Deserialize the returned descriptor |
| returned_descriptor = BlobDescriptor.deserialize(read_blob_bytes) |
| |
| # The returned descriptor should point to the blob file (simplified implementation) |
| # because the current implementation creates a descriptor pointing to the blob file location |
| self.assertEqual(returned_descriptor.uri, str(blob_file_path)) |
| self.assertGreater(returned_descriptor.offset, 0) # Should have some offset in the blob file |
| |
| reader.close() |
| |
| reader_content = FormatBlobReader( |
| file_io=file_io, |
| file_path=str(blob_file_path), |
| read_fields=[blob_field_name], |
| full_fields=read_fields, |
| push_down_predicate=None, |
| blob_as_descriptor=False |
| ) |
| batch_content = reader_content.read_arrow_batch() |
| self.assertIsNotNone(batch_content) |
| self.assertEqual(batch_content.num_rows, 1) |
| read_content_bytes = batch_content.column(0)[0].as_py() |
| self.assertIsInstance(read_content_bytes, bytes) |
| # When blob_as_descriptor=False, we should get the actual file content |
| self.assertEqual(read_content_bytes, test_content) |
| reader_content.close() |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
