YannByron commented on code in PR #8619:
URL: https://github.com/apache/paimon/pull/8619#discussion_r3578095468
##########
paimon-python/pypaimon/daft/daft_datasource.py:
##########
@@ -286,6 +286,80 @@ def _cast_blob_columns_to_file(rb: RecordBatch,
blob_column_names: set[str]) ->
return RecordBatch.from_pydict(columns)
+def _blob_native_covering_files(
+ files: list[DataFileMeta],
+ task_columns: list[str],
+ blob_column_names: set[str],
+ partition_keys: list[str],
+) -> list[DataFileMeta] | None:
+ """Return the parquet files that can serve a blob-table split via Daft's
+ native reader, or ``None`` if the split must use the pypaimon fallback.
+
+ A blob table stores each column bunch in its own file: scalar columns in
+ parquet, BLOB columns in ``.blob`` files, vector columns in ``.vector``
+ files, aligned by row id. Reading the base parquet files natively is only
+ correct when every projected data column lives in parquet files that each
+ fully cover the projection over disjoint row-id ranges -- i.e. no blob /
+ vector file carries a projected column, no cross-file field merge is
+ required, and no two covering files overlap. Partition columns are
+ path-derived, so they are excluded from file coverage. File-name
+ conventions mirror ``DataFileMeta.is_blob_file`` / ``is_vector_file``.
+ """
+ partitions = set(partition_keys)
+ projected = {c for c in task_columns if c not in partitions}
+ if projected & set(blob_column_names):
+ return None
+
+ covering: list[DataFileMeta] = []
+ for f in files:
+ name = f.file_name
+ write_cols = set(f.write_cols or [])
+ carried = write_cols & projected
+ if name.endswith(".blob") or ".vector." in name:
+ if carried:
+ return None # a projected column lives in a blob/vector bunch
+ continue
+ if not name.endswith(".parquet"):
+ return None # unknown bunch format; stay on the safe fallback path
+ if projected <= write_cols:
+ covering.append(f)
+ elif carried:
+ return None # partial coverage -> cross-file field merge required
+ # else: parquet file irrelevant to the projection -> skip it
+
+ if not covering:
+ return None
+
+ # The covering parquet files must not overlap (else rows are duplicated)
+ # and together must span every row-id range present in the split. Otherwise
+ # a row-id range whose projected column is absent from any covering file --
+ # e.g. an older data-evolution range written before the column existed --
+ # would be silently dropped here, whereas the pypaimon fallback returns
+ # those rows with the column read as null via schema evolution.
+ covering_ranges = []
+ for f in covering:
+ if f.first_row_id is None:
+ return None
+ covering_ranges.append((f.first_row_id, f.first_row_id + f.row_count))
+ covering_ranges.sort()
+ merged: list[tuple[int, int]] = []
+ for start, end in covering_ranges:
+ if merged and start < merged[-1][1]:
+ return None # overlapping row-id ranges would duplicate rows
+ if merged and start == merged[-1][1]:
+ merged[-1] = (merged[-1][0], end) # adjacent -> extend
+ else:
+ merged.append((start, end))
+
+ for f in files:
Review Comment:
I think here should pass over `covering`, not `files`. Correct me if I am
wrong.
--
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]
