RyanLin5967 opened a new issue, #10886: URL: https://github.com/apache/paimon/issues/10886
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/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. ```python 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! -- 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]
