Repository navigation
[Bug] [Python] Reads drop commits after rollback_to a tag whose snapshot expired #10886
Description
Search before asking
- I searched in the issues and found nothing similar.
Paimon version
pypaimon 2.1.0. rollback_helper.py is unchanged on master.
Compute Engine
Python API (pypaimon 2.1.0, Python 3.12.16, pyarrow 19.0.1) on a local filesystem catalog. Spark 3.5.6 with paimon-spark-3.5_2.12 2.1.0 on Java 17 expires the snapshots, since pypaimon has no snapshot expiration.
Minimal reproduce step
Currently, when the tag's snapshot has expired, pypaimon's rollback_to writes that snapshot file back but leaves the EARLIEST hint on the old earliest id, so once new commits reach that id, reads take it as the earliest snapshot and skip the commits before it. Spark's rollback procedure in its place sets EARLIEST to 2, and every read is right.
import os
import pyarrow as pa
from pyspark.sql import SparkSession
from pypaimon import CatalogFactory, Schema
spark = (SparkSession.builder
.config("spark.jars", "paimon-spark-3.5_2.12-2.1.0.jar")
.config("spark.sql.extensions", "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
.config("spark.sql.catalog.paimon", "org.apache.paimon.spark.SparkCatalog")
.config("spark.sql.catalog.paimon.warehouse", "file://" + os.path.abspath("wh"))
.getOrCreate())
cat = CatalogFactory.create({"warehouse": "wh"})
cat.create_database("db", True)
sch = pa.schema([("k", pa.int64())])
cat.create_table("db.t", Schema.from_pyarrow_schema(sch), False)
def write(k):
wb = cat.get_table("db.t").new_batch_write_builder()
w, c = wb.new_write(), wb.new_commit()
w.write_arrow(pa.table({"k": [k]}, schema=sch))
c.commit(w.prepare_commit())
def read(opts):
rb = cat.get_table("db.t").copy(opts).new_read_builder()
return sorted(rb.new_read().to_arrow(rb.new_scan().plan().splits())["k"].to_pylist())
for k in (1, 2, 3, 4, 5):
write(k)
cat.get_table("db.t").create_tag("v2", 2)
spark.sql("CALL paimon.sys.expire_snapshots(table => 'db.t', retain_max => 1, retain_min => 1)").collect()
cat.get_table("db.t").rollback_to("v2")
for k in (6, 7, 8):
write(k)
print(open("wh/db.db/t/snapshot/EARLIEST").read())
# 5, should be 2
sm = cat.get_table("db.t").snapshot_manager()
s2, s3, s5 = (sm.get_snapshot_by_id(i).time_millis for i in (2, 3, 5))
print(read({"incremental-between-timestamp": f"{s2},{s5}"}))
# [8], should be [6, 7, 8]
print(read({"scan.timestamp-millis": str(s3)}))
# ValueError: No snapshot found with timestamp earlier than or equal to 1793326991626ms., should be [1, 2, 6]What doesn't meet your expectations?
After rollback_to("v2") the EARLIEST hint is 2 and the reads return [6, 7, 8] and [1, 2, 6].
Anything else?
A new Spark session reads the same window as [8] too.
Are you willing to submit a PR?
- I'm willing to submit a PR!
Activity