This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1e059d4bec [python][daft] Fix data evolution fallback routing (#8670)
1e059d4bec is described below
commit 1e059d4bec7f6df20d03962a13200d1bb91180c6
Author: QuakeWang <[email protected]>
AuthorDate: Thu Jul 16 14:28:35 2026 +0800
[python][daft] Fix data evolution fallback routing (#8670)
Daft treated every append-table split as directly readable Parquet and
ignored `raw_convertible` for non-primary-key tables.
Non-raw-convertible data-evolution splits can contain column fragments
for the same row-id range, so reading each file independently returned
physical rows instead of merged logical rows.
Require `raw_convertible` for native Parquet routing, fall back to the
pypaimon merge reader otherwise, and report the correct data-evolution
fallback reason.
---
paimon-python/pypaimon/daft/daft_datasource.py | 6 ++-
.../pypaimon/tests/daft/daft_explain_test.py | 47 ++++++++++++++++++-
.../pypaimon/tests/daft/daft_integration_test.py | 52 ++++++++++++++++++++++
3 files changed, 102 insertions(+), 3 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_datasource.py
b/paimon-python/pypaimon/daft/daft_datasource.py
index f80ae62e5c..ccb98c1140 100644
--- a/paimon-python/pypaimon/daft/daft_datasource.py
+++ b/paimon-python/pypaimon/daft/daft_datasource.py
@@ -652,7 +652,7 @@ class PaimonDataSource(DataSource):
can_use_native_reader = (
self._is_parquet
and not self._has_blob_columns
- and (not self._table.is_primary_key_table or raw_convertible)
+ and raw_convertible
and not has_deletion_vectors
and not has_auth
)
@@ -667,8 +667,10 @@ class PaimonDataSource(DataSource):
reason = "query auth active"
elif has_deletion_vectors:
reason = "deletion vectors present"
- else:
+ elif self._table.is_primary_key_table:
reason = "LSM merge required"
+ else:
+ reason = "data-evolution merge required"
return _ReaderRouting(READER_MODE_PYPAIMON_FALLBACK, reason)
@staticmethod
diff --git a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
index f0941175dc..6f3a54a62b 100644
--- a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
@@ -90,6 +90,7 @@ def _single_split_explain(
raw_convertible: bool,
has_deletion_vectors: bool,
has_auth: bool = False,
+ data_evolution_enabled: bool = False,
) -> ExplainResult:
split = ExplainSplitInfo(
partition={},
@@ -109,7 +110,7 @@ def _single_split_explain(
is_primary_key_table=False,
bucket_mode="unaware",
deletion_vectors_enabled=has_deletion_vectors,
- data_evolution_enabled=False,
+ data_evolution_enabled=data_evolution_enabled,
snapshot_id=1,
schema_id=0,
file_count=1,
@@ -334,6 +335,50 @@ def
test_explain_scan_reports_pk_lsm_fallback(catalog_options):
assert all(split.fallback_reason == "LSM merge required" for split in
result.splits)
+def test_explain_scan_reports_data_evolution_fallback(catalog_options,
monkeypatch):
+ pa_schema = pa.schema([
+ ("id", pa.int64()),
+ ("name", pa.string()),
+ ])
+ _, table = _create_table(
+ catalog_options,
+ "explain_data_evolution_fallback",
+ pa_schema,
+ options={
+ "bucket": "-1",
+ "file.format": "parquet",
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ },
+ )
+
+ class FakeReadBuilder:
+ def explain(self, verbose: bool = False) -> ExplainResult:
+ assert verbose is True
+ return _single_split_explain(
+ table_identifier="test_db.explain_data_evolution_fallback",
+ raw_convertible=False,
+ has_deletion_vectors=False,
+ data_evolution_enabled=True,
+ )
+
+ def fake_scan_read_builder(self, table, read_pushdowns):
+ return FakeReadBuilder()
+
+ monkeypatch.setattr(PaimonDataSource, "_scan_read_builder",
fake_scan_read_builder)
+
+ result = _explain_table(table, catalog_options=catalog_options,
verbose=True)
+
+ assert table.is_primary_key_table is False
+ assert result.pypaimon_fallback_split_count == 1
+ assert result.native_parquet_split_count == 0
+ assert result.fallback_reasons == {"data-evolution merge required": 1}
+ assert result.splits is not None
+ assert len(result.splits) == 1
+ assert result.splits[0].reader_mode == READER_MODE_PYPAIMON_FALLBACK
+ assert result.splits[0].fallback_reason == "data-evolution merge required"
+
+
def test_explain_scan_reports_non_parquet_fallback(catalog_options):
pa_schema = pa.schema([
("id", pa.int64()),
diff --git a/paimon-python/pypaimon/tests/daft/daft_integration_test.py
b/paimon-python/pypaimon/tests/daft/daft_integration_test.py
index 4b1ff9fbe4..0a8e44466c 100644
--- a/paimon-python/pypaimon/tests/daft/daft_integration_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_integration_test.py
@@ -112,6 +112,58 @@ def test_read_paimon_basic(catalog_options):
}
+def test_read_paimon_data_evolution_merges_column_fragments(catalog_options):
+ pa_schema = pa.schema([
+ ("id", pa.int32()),
+ ("name", pa.string()),
+ ("score", pa.float64()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_data_evolution",
+ pa_schema,
+ options={
+ "row-tracking.enabled": "true",
+ "data-evolution.enabled": "true",
+ "file.format": "parquet",
+ },
+ )
+ write_builder = table.new_batch_write_builder()
+ id_name_write = write_builder.new_write().with_write_type(["id", "name"])
+ score_write = write_builder.new_write().with_write_type(["score"])
+ table_commit = write_builder.new_commit()
+ try:
+ id_name_write.write_arrow(pa.table({
+ "id": pa.array([1, 2, 3], pa.int32()),
+ "name": pa.array(["a", "b", "c"], pa.string()),
+ }))
+ score_write.write_arrow(pa.table({
+ "score": pa.array([1.1, 2.2, 3.3], pa.float64()),
+ }))
+ commit_messages = id_name_write.prepare_commit() +
score_write.prepare_commit()
+ # Both files are column fragments for the same logical row range.
+ for message in commit_messages:
+ for data_file in message.new_files:
+ data_file.first_row_id = 0
+ table_commit.commit(commit_messages)
+ finally:
+ id_name_write.close()
+ score_write.close()
+ table_commit.close()
+
+ splits = table.new_read_builder().new_scan().plan().splits()
+ assert len(splits) == 1
+ assert splits[0].raw_convertible is False
+
+ result = read_paimon(identifier, catalog_options).to_pydict()
+
+ assert result == {
+ "id": [1, 2, 3],
+ "name": ["a", "b", "c"],
+ "score": [1.1, 2.2, 3.3],
+ }
+
+
def test_read_paimon_projection(catalog_options):
data = pa.table(
{
