[python] Support blob read && write (#6420)

Author: leaves12138

paimon-python/pypaimon/common/core_options.py

@@ -49,3 +49,5 @@ def __str__(self):
     INCREMENTAL_BETWEEN_TIMESTAMP = "incremental-between-timestamp"
     # Commit options
     COMMIT_USER_PREFIX = "commit.user-prefix"
+    ROW_TRACKING_ENABLED = "row-tracking.enabled"
+    DATA_EVOLUTION_ENABLED = "data-evolution.enabled"

paimon-python/pypaimon/common/file_io.py

@@ -25,7 +25,6 @@
 import pyarrow
 from packaging.version import parse
 from pyarrow._fs import FileSystem
-
 from pypaimon.common.config import OssOptions, S3Options
 from pypaimon.common.uri_reader import UriReaderFactory
 from pypaimon.schema.data_types import DataField, AtomicType, PyarrowFieldParser

paimon-python/pypaimon/read/reader/concat_batch_reader.py

@@ -19,6 +19,7 @@
 import collections
 from typing import Callable, List, Optional
 
+import pyarrow as pa
 from pyarrow import RecordBatch
 
 from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
@@ -76,3 +77,70 @@ def read_arrow_batch(self) -> Optional[RecordBatch]:
                 return batch.slice(0, self.split_end_row - cur_begin)
         else:
             return batch
+
+
+class MergeAllBatchReader(RecordBatchReader):
+    """
+    A reader that accepts multiple reader suppliers and concatenates all their arrow batches
+    into one big batch. This is useful when you want to merge all data from multiple sources
+    into a single batch for processing.
+    """
+
+    def __init__(self, reader_suppliers: List[Callable]):
+        self.reader_suppliers = reader_suppliers
+        self.merged_batch: Optional[RecordBatch] = None
+        self.batch_created = False
+
+    def read_arrow_batch(self) -> Optional[RecordBatch]:
+        if self.batch_created:
+            return None
+
+        all_batches = []
+
+        # Read all batches from all reader suppliers
+        for supplier in self.reader_suppliers:
+            reader = supplier()
+            try:
+                while True:
+                    batch = reader.read_arrow_batch()
+                    if batch is None:
+                        break
+                    all_batches.append(batch)
+            finally:
+                reader.close()
+
+        # Concatenate all batches into one big batch
+        if all_batches:
+            # For PyArrow < 17.0.0, use Table.concat_tables approach
+            # Convert batches to tables and concatenate
+            tables = [pa.Table.from_batches([batch]) for batch in all_batches]
+            if len(tables) == 1:
+                # Single table, just get the first batch
+                self.merged_batch = tables[0].to_batches()[0]
+            else:
+                # Multiple tables, concatenate them
+                concatenated_table = pa.concat_tables(tables)
+                # Convert back to a single batch by taking all batches and combining
+                all_concatenated_batches = concatenated_table.to_batches()
+                if len(all_concatenated_batches) == 1:
+                    self.merged_batch = all_concatenated_batches[0]
+                else:
+                    # If still multiple batches, we need to manually combine them
+                    # This shouldn't happen with concat_tables, but just in case
+                    combined_arrays = []
+                    for i in range(len(all_concatenated_batches[0].columns)):
+                        column_arrays = [batch.column(i) for batch in all_concatenated_batches]
+                        combined_arrays.append(pa.concat_arrays(column_arrays))
+                    self.merged_batch = pa.RecordBatch.from_arrays(
+                        combined_arrays,
+                        names=all_concatenated_batches[0].schema.names
+                    )
+        else:
+            self.merged_batch = None
+
+        self.batch_created = True
+        return self.merged_batch
+
+    def close(self) -> None:
+        self.merged_batch = None
+        self.batch_created = False

paimon-python/pypaimon/read/scanner/full_starting_scanner.py

@@ -18,6 +18,7 @@
 from collections import defaultdict
 from typing import Callable, List, Optional
 
+from pypaimon.common.core_options import CoreOptions
 from pypaimon.common.predicate import Predicate
 from pypaimon.common.predicate_builder import PredicateBuilder
 from pypaimon.manifest.manifest_file_manager import ManifestFileManager
@@ -65,13 +66,16 @@ def __init__(self, table, predicate: Optional[Predicate], limit: Optional[int],
 
         self.only_read_real_buckets = True if int(
             self.table.options.get('bucket', -1)) == BucketMode.POSTPONE_BUCKET.value else False
+        self.data_evolution = self.table.options.get(CoreOptions.DATA_EVOLUTION_ENABLED, 'false').lower() == 'true'
 
     def scan(self) -> Plan:
         file_entries = self.plan_files()
         if not file_entries:
             return Plan([])
         if self.table.is_primary_key_table:
             splits = self._create_primary_key_splits(file_entries)
+        elif self.data_evolution:
+            splits = self._create_data_evolution_splits(file_entries)
         else:
             splits = self._create_append_only_splits(file_entries)
 
@@ -104,7 +108,7 @@ def plan_files(self) -> List[ManifestEntry]:
             file_entries = self._filter_by_predicate(file_entries)
         return file_entries
 
-    def with_shard(self, idx_of_this_subtask, number_of_para_subtasks) -> 'TableScan':
+    def with_shard(self, idx_of_this_subtask, number_of_para_subtasks) -> 'FullStartingScanner':
         if idx_of_this_subtask >= number_of_para_subtasks:
             raise Exception("idx_of_this_subtask must be less than number_of_para_subtasks")
         self.idx_of_this_subtask = idx_of_this_subtask
@@ -357,3 +361,114 @@ def _pack_for_ordered(items: List, weight_func: Callable, target_weight: int) ->
             packed.append(bin_items)
 
         return packed
+
+    def _create_data_evolution_splits(self, file_entries: List[ManifestEntry]) -> List['Split']:
+        partitioned_files = defaultdict(list)
+        for entry in file_entries:
+            partitioned_files[(tuple(entry.partition.values), entry.bucket)].append(entry)
+
+        if self.idx_of_this_subtask is not None:
+            partitioned_files, plan_start_row, plan_end_row = self._append_only_filter_by_shard(partitioned_files)
+
+        def weight_func(file_list: List[DataFileMeta]) -> int:
+            return max(sum(f.file_size for f in file_list), self.open_file_cost)
+
+        splits = []
+        for key, file_entries in partitioned_files.items():
+            if not file_entries:
+                continue
+
+            data_files: List[DataFileMeta] = [e.file for e in file_entries]
+
+            # Split files by firstRowId for data evolution
+            split_by_row_id = self._split_by_row_id(data_files)
+
+            # Pack the split groups for optimal split sizes
+            packed_files: List[List[List[DataFileMeta]]] = self._pack_for_ordered(split_by_row_id, weight_func,
+                                                                                  self.target_split_size)
+
+            # Flatten the packed files and build splits
+            flatten_packed_files: List[List[DataFileMeta]] = [
+                [file for sub_pack in pack for file in sub_pack]
+                for pack in packed_files
+            ]
+
+            splits += self._build_split_from_pack(flatten_packed_files, file_entries, False)
+
+        if self.idx_of_this_subtask is not None:
+            self._compute_split_start_end_row(splits, plan_start_row, plan_end_row)
+        return splits
+
+    def _split_by_row_id(self, files: List[DataFileMeta]) -> List[List[DataFileMeta]]:
+        split_by_row_id = []
+
+        def sort_key(file: DataFileMeta) -> tuple:
+            first_row_id = file.first_row_id if file.first_row_id is not None else float('-inf')
+            is_blob = 1 if self._is_blob_file(file.file_name) else 0
+            # For files with same firstRowId, sort by maxSequenceNumber in descending order
+            # (larger sequence number means more recent data)
+            max_seq = file.max_sequence_number
+            return (first_row_id, is_blob, -max_seq)
+
+        sorted_files = sorted(files, key=sort_key)
+
+        # Filter blob files to only include those within the row ID range of non-blob files
+        sorted_files = self._filter_blob(sorted_files)
+
+        # Split files by firstRowId
+        last_row_id = -1
+        check_row_id_start = 0
+        current_split = []
+
+        for file in sorted_files:
+            first_row_id = file.first_row_id
+            if first_row_id is None:
+                # Files without firstRowId are treated as individual splits
+                split_by_row_id.append([file])
+                continue
+
+            if not self._is_blob_file(file.file_name) and first_row_id != last_row_id:
+                if current_split:
+                    split_by_row_id.append(current_split)
+
+                # Validate that files don't overlap
+                if first_row_id < check_row_id_start:
+                    file_names = [f.file_name for f in sorted_files]
+                    raise ValueError(
+                        f"There are overlapping files in the split: {file_names}, "
+                        f"the wrong file is: {file.file_name}"
+                    )
+
+                current_split = []
+                last_row_id = first_row_id
+                check_row_id_start = first_row_id + file.row_count
+
+            current_split.append(file)
+
+        if current_split:
+            split_by_row_id.append(current_split)
+
+        return split_by_row_id
+
+    @staticmethod
+    def _is_blob_file(file_name: str) -> bool:
+        return file_name.endswith('.blob')
+
+    @staticmethod
+    def _filter_blob(files: List[DataFileMeta]) -> List[DataFileMeta]:
+        result = []
+        row_id_start = -1
+        row_id_end = -1
+
+        for file in files:
+            if not FullStartingScanner._is_blob_file(file.file_name):
+                if file.first_row_id is not None:
+                    row_id_start = file.first_row_id
+                    row_id_end = file.first_row_id + file.row_count
+                result.append(file)
+            else:
+                if file.first_row_id is not None and row_id_start != -1:
+                    if row_id_start <= file.first_row_id < row_id_end:
+                        result.append(file)
+
+        return result

paimon-python/pypaimon/read/split_read.py

@@ -26,7 +26,7 @@
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.read.interval_partition import IntervalPartition, SortedRun
 from pypaimon.read.partition_info import PartitionInfo
-from pypaimon.read.reader.concat_batch_reader import ConcatBatchReader, ShardBatchReader
+from pypaimon.read.reader.concat_batch_reader import ConcatBatchReader, ShardBatchReader, MergeAllBatchReader
 from pypaimon.read.reader.concat_record_reader import ConcatRecordReader
 from pypaimon.read.reader.data_file_batch_reader import DataFileBatchReader
 from pypaimon.read.reader.data_evolution_merge_reader import DataEvolutionMergeReader
@@ -73,21 +73,21 @@ def __init__(self, table, predicate: Optional[Predicate], push_down_predicate,
     def create_reader(self) -> RecordReader:
         """Create a record reader for the given split."""
 
-    def file_reader_supplier(self, file_path: str, for_merge_read: bool):
+    def file_reader_supplier(self, file_path: str, for_merge_read: bool, read_fields: List[str]):
         _, extension = os.path.splitext(file_path)
         file_format = extension[1:]
 
         format_reader: RecordBatchReader
         if file_format == CoreOptions.FILE_FORMAT_AVRO:
-            format_reader = FormatAvroReader(self.table.file_io, file_path, self._get_final_read_data_fields(),
+            format_reader = FormatAvroReader(self.table.file_io, file_path, read_fields,
                                              self.read_fields, self.push_down_predicate)
         elif file_format == CoreOptions.FILE_FORMAT_BLOB:
             blob_as_descriptor = self.table.options.get(CoreOptions.FILE_BLOB_AS_DESCRIPTOR, False)
-            format_reader = FormatBlobReader(self.table.file_io, file_path, self._get_final_read_data_fields(),
+            format_reader = FormatBlobReader(self.table.file_io, file_path, read_fields,
                                              self.read_fields, self.push_down_predicate, blob_as_descriptor)
         elif file_format == CoreOptions.FILE_FORMAT_PARQUET or file_format == CoreOptions.FILE_FORMAT_ORC:
             format_reader = FormatPyArrowReader(self.table.file_io, file_format, file_path,
-                                                self._get_final_read_data_fields(), self.push_down_predicate)
+                                                read_fields, self.push_down_predicate)
         else:
             raise ValueError(f"Unexpected file format: {file_format}")
 
@@ -253,7 +253,12 @@ class RawFileSplitRead(SplitRead):
     def create_reader(self) -> RecordReader:
         data_readers = []
         for file_path in self.split.file_paths:
-            supplier = partial(self.file_reader_supplier, file_path=file_path, for_merge_read=False)
+            supplier = partial(
+                self.file_reader_supplier,
+                file_path=file_path,
+                for_merge_read=False,
+                read_fields=self._get_final_read_data_fields(),
+            )
             data_readers.append(supplier)
 
         if not data_readers:
@@ -274,7 +279,12 @@ def _get_all_data_fields(self):
 
 class MergeFileSplitRead(SplitRead):
     def kv_reader_supplier(self, file_path):
-        reader_supplier = partial(self.file_reader_supplier, file_path=file_path, for_merge_read=True)
+        reader_supplier = partial(
+            self.file_reader_supplier,
+            file_path=file_path,
+            for_merge_read=True,
+            read_fields=self._get_final_read_data_fields()
+        )
         return KeyValueWrapReader(reader_supplier(), len(self.trimmed_primary_key), self.value_arity)
 
     def section_reader_supplier(self, section: List[SortedRun]):
@@ -317,7 +327,7 @@ def create_reader(self) -> RecordReader:
             if len(need_merge_files) == 1 or not self.read_fields:
                 # No need to merge fields, just create a single file reader
                 suppliers.append(
-                    lambda f=need_merge_files[0]: self._create_file_reader(f)
+                    lambda f=need_merge_files[0]: self._create_file_reader(f, self._get_final_read_data_fields())
                 )
             else:
                 suppliers.append(
@@ -424,26 +434,30 @@ def _create_union_reader(self, need_merge_files: List[DataFileMeta]) -> RecordRe
                 self.read_fields = read_fields  # create reader based on read_fields
                 # Create reader for this bunch
                 if len(bunch.files()) == 1:
-                    file_record_readers[i] = self._create_file_reader(bunch.files()[0])
+                    file_record_readers[i] = self._create_file_reader(
+                        bunch.files()[0], [field.name for field in read_fields]
+                    )
                 else:
                     # Create concatenated reader for multiple files
                     suppliers = [
-                        lambda f=file: self._create_file_reader(f) for file in bunch.files()
+                        lambda f=file: self._create_file_reader(
+                            f, [field.name for field in read_fields]
+                        ) for file in bunch.files()
                     ]
-                    file_record_readers[i] = ConcatRecordReader(suppliers)
+                    file_record_readers[i] = MergeAllBatchReader(suppliers)
                 self.read_fields = table_fields
 
         # Validate that all required fields are found
         for i, field in enumerate(all_read_fields):
             if row_offsets[i] == -1:
-                if not field.type.is_nullable():
+                if not field.type.nullable:
                     raise ValueError(f"Field {field} is not null but can't find any file contains it.")
 
         return DataEvolutionMergeReader(row_offsets, field_offsets, file_record_readers)
 
-    def _create_file_reader(self, file: DataFileMeta) -> RecordReader:
+    def _create_file_reader(self, file: DataFileMeta, read_fields: [str]) -> RecordReader:
         """Create a file reader for a single file."""
-        return self.file_reader_supplier(file_path=file.file_path, for_merge_read=False)
+        return self.file_reader_supplier(file_path=file.file_path, for_merge_read=False, read_fields=read_fields)
 
     def _split_field_bunches(self, need_merge_files: List[DataFileMeta]) -> List[FieldBunch]:
         """Split files into field bunches."""

paimon-python/pypaimon/read/table_read.py

@@ -20,6 +20,7 @@
 import pandas
 import pyarrow
 
+from pypaimon.common.core_options import CoreOptions
 from pypaimon.common.predicate import Predicate
 from pypaimon.common.predicate_builder import PredicateBuilder
 from pypaimon.read.push_down_utils import extract_predicate_to_list
@@ -132,7 +133,7 @@ def _create_split_read(self, split: Split) -> SplitRead:
                 read_type=self.read_type,
                 split=split
             )
-        elif self.table.options.get('data-evolution.enabled', 'false').lower() == 'true':
+        elif self.table.options.get(CoreOptions.DATA_EVOLUTION_ENABLED, 'false').lower() == 'true':
             return DataEvolutionSplitRead(
                 table=self.table,
                 predicate=self.predicate,

paimon-python/pypaimon/schema/data_types.py

@@ -548,6 +548,7 @@ def to_paimon_field(field_idx: int, pa_field: pyarrow.Field) -> DataField:
 
     @staticmethod
     def to_paimon_schema(pa_schema: pyarrow.Schema) -> List[DataField]:
+        # Convert PyArrow schema to Paimon fields
         fields = []
         for i, pa_field in enumerate(pa_schema):
             pa_field: pyarrow.Field