clean code

Author: XiaoHongbo-Hope

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

@@ -210,28 +210,19 @@ def read_arrow_batch(self) -> Optional[RecordBatch]:
             batch_index = self.row_offsets[i]
             field_index = self.field_offsets[i]
             field_name = self.schema.field(i).name
-            column = None
 
             if batch_index >= 0 and batches[batch_index] is not None:
                 src_batch = batches[batch_index]
-                if field_name is not None and field_name in src_batch.schema.names:
+                if field_name in src_batch.schema.names:
                     column = src_batch.column(
                         src_batch.schema.get_field_index(field_name)
                     ).slice(0, min_rows)
-                elif field_index < src_batch.num_columns:
-                    column = src_batch.column(field_index).slice(0, min_rows)
-
-            if column is None and field_name is not None:
-                for b in batches:
-                    if b is not None and field_name in b.schema.names:
-                        column = b.column(b.schema.get_field_index(field_name)).slice(
-                            0, min_rows
-                        )
-                        break
-
-            if column is not None:
-                columns.append(column)
-            elif self.schema is not None and i < len(self.schema):
+                    columns.append(column)
+                else:
+                    # Field doesn't exist in this batch, fill with nulls
+                    columns.append(pa.nulls(min_rows, type=self.schema.field(i).type))
+            else:
+                # No batch provides this field, fill with nulls
                 columns.append(pa.nulls(min_rows, type=self.schema.field(i).type))
 
         for i in range(len(self.readers)):

paimon-python/pypaimon/read/split_read.py

@@ -806,15 +806,6 @@ def _use_requested_field_names_for_merge(self, fields_files: List[FieldBunch]) -
     def _create_file_reader(self, file: DataFileMeta, read_fields: [str],
                             use_requested_field_names: bool = True) -> Optional[RecordReader]:
         """Create a file reader for a single file."""
-        shard_file_idx_map = (
-            self.split.shard_file_idx_map() if isinstance(self.split, SlicedSplit) else {}
-        )
-        begin_pos, end_pos = 0, 0
-        if file.file_name in shard_file_idx_map:
-            (begin_pos, end_pos) = shard_file_idx_map[file.file_name]
-            if (begin_pos, end_pos) == (-1, -1):
-                return None
-
         def create_record_reader():
             return self.file_reader_supplier(
                 file=file,
@@ -824,8 +815,6 @@ def create_record_reader():
                 use_requested_field_names=use_requested_field_names)
 
         base = create_record_reader()
-        if file.file_name in shard_file_idx_map:
-            base = ShardBatchReader(base, begin_pos, end_pos)
         if self.row_ranges is None:
             return base
         file_range = file.row_id_range()

paimon-python/pypaimon/read/table_read.py

@@ -68,8 +68,6 @@ def _try_to_pad_batch_by_schema(batch: pyarrow.RecordBatch, target_schema):
         for field in target_schema:
             if field.name in batch.schema.names:
                 col = batch.column(field.name)
-                if col.type != field.type:
-                    col = pyarrow.nulls(num_rows, type=field.type)
             else:
                 col = pyarrow.nulls(num_rows, type=field.type)
             columns.append(col)