You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "TheR1sing3un (via GitHub)" <gi...@apache.org> on 2026/10/08 06:10:41 UTC
[PR] [python] Add atomic global index maintenance and replacement [paimon]
TheR1sing3un opened a new pull request, #10431:
URL: https://github.com/apache/paimon/pull/10431
### Purpose
Add opt-in `maintain_global_indexes` / multimodal `maintain_indexes` for configured global indexes. Each pass plans one snapshot, fills uncovered ranges, and publishes all requested definitions in one commit. Repeated catch-up is a no-op; keeping explicit definitions lets scheduled ingestion/backfill jobs rebuild coverage removed by `DROP_PARTITION_INDEX` updates.
`rebuild=True` atomically replaces matching index entries, optionally for selected partitions, without removing files retained by older snapshots. A strict Python-committer snapshot guard rejects publication after any concurrent commit. Failed builds preserve existing coverage and clean completed uncommitted outputs; ambiguous commit failures leave files for orphan maintenance. The API does not create a background service: callers own scheduling, backoff and retries. `IGNORE` updates require explicit rebuild.
### Tests
83 targeted maintenance/build/conflict tests passed across the initial 81-test run and the final expanded 10-test maintenance run. Coverage includes idempotence, append catch-up, rebuilding update-invalidated partitions, one-commit publication of multiple definitions, partition-scoped replacement, historical index retention, concurrent-commit rejection, failed-build cleanup, ignored-update repair and validation. Flake8 and `git diff --check` passed. No benchmark code included.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Add atomic global index maintenance and replacement [paimon]
Posted by "TheR1sing3un (via GitHub)" <gi...@apache.org>.
TheR1sing3un closed pull request #10431: [python] Add atomic global index maintenance and replacement
URL: https://github.com/apache/paimon/pull/10431
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Add atomic global index maintenance and replacement [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10431:
URL: https://github.com/apache/paimon/pull/10431#discussion_r4253689831
##########
docs/docs/pypaimon/multimodal-search.md:
##########
@@ -145,6 +145,53 @@ matches = (
)
```
+## Maintain index coverage and rebuild indexes
+
+Keep the desired index definitions in your ingestion/backfill job and call
+`maintain_indexes` after committing data, or from a scheduled job:
+
+```python
+indexes = [{"column": "embedding", "type": "ivf-flat", "options": {
+ "ivf-flat.nlist": "128", "global-index.row-count-per-shard": "1000000",
+}}]
+added_files = docs.maintain_indexes(indexes)
+```
+
+Each pass automatically discovers uncovered ranges and builds them. Repeating a
+completed pass is a no-op. For embedding updates, configure the table with
+`global-index.column-update-action=DROP_PARTITION_INDEX`: the data commit removes
+stale coverage, and a later maintenance pass rebuilds the affected partitions.
+Keep query search mode `full` if queries must include rows before that rebuild.
+Definitions must remain in your job configuration because dropping the last
+index file also removes the manifest information needed to rediscover its type.
+
+To change build parameters or repair coverage left stale by `IGNORE`, use an
+explicit rebuild, optionally scoped to partitions:
+
+```python
+docs.maintain_indexes(indexes, rebuild=True, partitions={"date": "2026-10-08"})
+```
+
+All requested indexes are built against one captured snapshot and published in
+one commit. Rebuild deletes the matching old entries and adds replacements in
+that same commit; old files remain available to retained snapshots. Any
+concurrent snapshot commit causes publication to fail. Every publication attempt
+checks the captured snapshot metadata, including its UUID and manifests, so a
+rollback followed by reuse of the same snapshot ID also invalidates the plan.
Review Comment:
[P2] Qualify the rollback guarantee for filesystem publication
The new captured-snapshot check rejects changes before conflict checking and on retries, but this unconditional rollback guarantee is broader than the filesystem committer can provide. RenamingSnapshotCommit ignores base_snapshot_uuid and has no shared lock with rollback. With real Parquet/BTree data, I interleaved public rollback_to(1) and append(20) at the final commit entry, then delegated to the unmodified production commit: replacement snapshot 2 contained [0,20], but maintenance successfully published snapshot 3 containing the old [0,10]. No false commit result was injected. This is an existing shared committer limitation, so please document that rollback on such catalogs requires pausing/draining maintenance publication as well as other writers, or limit this guarantee to a committer that atomically enforces the captured identity. A second preflight read alone would not make it atomic.
##########
paimon-python/pypaimon/tests/global_index_maintenance_test.py:
##########
@@ -0,0 +1,327 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+import pypaimon.multimodal as pm
+from pypaimon.globalindex.create_global_index import GlobalIndexBuilder
+from pypaimon.index.index_file_handler import IndexFileHandler
+
+SPEC = {"column": "embedding", "type": "ivf-flat", "options": {
+ "ivf-flat.nlist": "1", "global-index.row-count-per-shard": "2"}}
+
+
+@pytest.fixture
+def docs(tmp_path):
+ pytest.importorskip("paimon_vindex")
+ table = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "docs", schema=pa.schema([("id", pa.int64()), ("pt", pa.string()),
+ ("embedding", pa.list_(pa.float32(), 2))]), partitioned=["pt"],
+ options={"file.format": "parquet", "vector.file.format": "parquet",
+ "vector-index.search-mode": "full", "global-index.column-update-action": "DROP_PARTITION_INDEX"})
+ table.add([{"id": i, "pt": "a" if i < 2 else "b", "embedding": [float(i), 1.]}
+ for i in range(4)])
+ return table
+
+
+def entries(docs, snapshot=None):
+ table = docs.raw_table
+ return IndexFileHandler(table).scan(snapshot or table.snapshot_manager().get_latest_snapshot())
+
+
+def names(docs, snapshot=None):
+ return {entry.index_file.file_name for entry in entries(docs, snapshot)}
+
+
+def test_catch_up_is_idempotent_and_restores_dropped_partition(docs):
+ assert docs.maintain_indexes([SPEC]) == 2
+ original = names(docs)
+ snapshot = docs.raw_table.snapshot_manager().get_latest_snapshot()
+ assert docs.maintain_indexes([SPEC]) == 0
+ assert docs.raw_table.snapshot_manager().get_latest_snapshot().id == snapshot.id
+ docs.add([{"id": 4, "pt": "b", "embedding": [4., 1.]}])
+ assert docs.maintain_indexes([SPEC]) == 1
+ assert names(docs).issuperset(original)
+ docs.update("id = 0", {"embedding": [100., 1.]})
+ preserved = names(docs)
+ assert len(preserved) == 2 # Unmodified partition b keeps both files.
+ assert docs.maintain_indexes([SPEC]) == 1
+ assert names(docs).issuperset(preserved)
+ assert docs.search([100., 1.]).select(["id"]).limit(1).to_list() == [{"id": 0}]
+
+
+def test_partition_replacement_is_atomic_and_old_snapshot_remains_readable(docs):
+ docs.maintain_indexes([SPEC])
+ before = docs.raw_table.snapshot_manager().get_latest_snapshot()
+ old = names(docs)
+ untouched = {entry.index_file.file_name for entry in entries(docs) if entry.partition.values == ["b"]}
+ assert docs.maintain_indexes([SPEC], rebuild=True, partitions={"pt": "a"}) == 1
+ assert docs.raw_table.snapshot_manager().get_latest_snapshot().id == before.id + 1
+ assert names(docs) & old == untouched
+ assert names(docs, before) == old
+ for name in old:
+ assert docs.raw_table.file_io.exists(docs.raw_table.path_factory().global_index_path_factory().to_path(name))
+ # Inner product with [0, 1] ties every row. Row IDs depend on partition
+ # write order, so use a query whose unique best match is id=0.
+ assert docs.search([-1., 1.], snapshot_id=before.id).select(["id"]).limit(1).to_list() == [{"id": 0}]
+
+
+def test_concurrent_commit_rejects_replacement(docs):
+ docs.maintain_indexes([SPEC])
+ old = names(docs)
+ original = GlobalIndexBuilder.build
+
+ def append_after_build(builder):
+ messages = original(builder)
+ docs.add([{"id": 5, "pt": "a", "embedding": [5., 1.]}])
+ return messages
+
+ with patch.object(GlobalIndexBuilder, "build", append_after_build):
+ with pytest.raises(RuntimeError, match="Global index maintenance conflict"):
+ docs.maintain_indexes([SPEC], rebuild=True)
+ assert names(docs) == old
+ assert docs.scan().to_arrow().num_rows == 5
+ assert docs.maintain_indexes([SPEC], rebuild=True) == 3
+
+
+@pytest.mark.parametrize("rebuild", [False, True])
+@pytest.mark.parametrize("during_retry", [False, True])
+@pytest.mark.parametrize("legacy_snapshot", [False, True])
+def test_rollback_and_reappend_rejects_reused_snapshot(tmp_path, rebuild, during_retry, legacy_snapshot):
+ from pypaimon.common.json_util import JSON
+ from pypaimon.globalindex.data_evolution_global_index_scanner import DataEvolutionGlobalIndexScanner
+ from pypaimon.snapshot.renaming_snapshot_commit import RenamingSnapshotCommit
+
+ docs = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "docs", schema=pa.schema([("id", pa.int64())]),
+ options={"file.format": "parquet", "write.native.enabled": "false",
+ "commit.native.enabled": "false"})
+ docs.add([{"id": 0}])
+ spec = {"column": "id", "type": "btree"}
+ if rebuild:
+ docs.maintain_indexes([spec])
+ table = docs.raw_table
+
+ def latest_snapshot():
+ manager = table.snapshot_manager()
+ snapshot = manager.get_latest_snapshot()
+ if legacy_snapshot:
+ # Older snapshot files have no UUID. Preserve that representation
+ # for both versions of the reused snapshot ID.
+ snapshot.uuid = None
+ table.file_io.write_file(manager.get_snapshot_path(snapshot.id),
+ JSON.to_json(snapshot), overwrite=True)
+ return snapshot
+
+ rollback_snapshot = table.snapshot_manager().get_latest_snapshot()
+ old_indexes = names(docs)
+ docs.add([{"id": 10}])
+ planned = latest_snapshot()
+ replacements = []
+
+ def replace_snapshot():
+ table.rollback_to(rollback_snapshot.id)
+ docs.add([{"id": 20}])
+ replacement = latest_snapshot()
+ assert (replacement.id, replacement.schema_id, replacement.next_row_id) == (
+ planned.id, planned.schema_id, planned.next_row_id)
+ if legacy_snapshot:
+ assert replacement.uuid is planned.uuid is None
+ else:
+ assert replacement.uuid != planned.uuid
+ assert replacement.delta_manifest_list != planned.delta_manifest_list
+ replacements.append(replacement)
+
+ original_build = GlobalIndexBuilder.build
+ original_commit = RenamingSnapshotCommit.commit
+
+ def build(builder):
+ messages = original_build(builder)
+ if not during_retry:
+ replace_snapshot()
+ return messages
+
+ def commit(committer, previous_uuid, snapshot, statistics):
+ if during_retry and snapshot.index_manifest and not replacements:
+ # Simulate a failed CAS after the first check; the retry must
+ # still compare against the original plan, not the reused ID.
+ with patch.object(RenamingSnapshotCommit, "commit", original_commit):
+ replace_snapshot()
+ return False
+ return original_commit(committer, previous_uuid, snapshot, statistics)
+
+ with patch.object(GlobalIndexBuilder, "build", build), \
+ patch.object(RenamingSnapshotCommit, "commit", commit):
+ with pytest.raises(RuntimeError, match="Global index maintenance conflict"):
+ docs.maintain_indexes([spec], rebuild=rebuild)
+
+ assert len(replacements) == 1
+ assert table.snapshot_manager().get_latest_snapshot() == replacements[0]
+ assert names(docs) == old_indexes
+ assert docs.scan().to_arrow().sort_by("id").to_pylist() == [{"id": 0}, {"id": 20}]
+ assert docs.scan().where("id = 20").to_list() == [{"id": 20}]
+ assert docs.maintain_indexes([spec], rebuild=rebuild) > 0
+ predicate = table.new_read_builder().new_predicate_builder().equal("id", 20)
+ with DataEvolutionGlobalIndexScanner.create(
+ table, predicate=predicate,
+ snapshot=table.snapshot_manager().get_latest_snapshot()) as scanner:
+ result = scanner.scan(predicate)
+ assert result is not None
+ rb = table.new_read_builder().with_filter(predicate)
+ splits = rb.new_scan().with_global_index_result(result).plan().splits()
+ assert rb.new_read().to_arrow(splits).to_pylist() == [{"id": 20}]
+
+
+@pytest.mark.parametrize("rebuild", [False, True])
+def test_concurrent_compaction_is_not_rolled_back(tmp_path, rebuild):
+ docs = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "docs", schema=pa.schema([("id", pa.int64())]),
+ options={"file.format": "parquet", "commit.native.enabled": "false",
+ "write.native.enabled": "false"})
+ docs.add([{"id": 0}, {"id": 1}])
+ spec = {"column": "id", "type": "btree"}
+ if rebuild:
+ docs.maintain_indexes([spec])
+ table = docs.raw_table
+ old_indexes = names(docs)
+ old_files = {file.file_name for split in table.new_read_builder().new_scan().plan().splits()
+ for file in split.files}
+ original = GlobalIndexBuilder.build
+ compacted = []
+
+ def compact_after_build(builder):
+ messages = original(builder)
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().plan_for_write().splits()
+ current = read_builder.new_read().to_arrow(splits).sort_by("id")
+ wb = table.new_batch_write_builder()
+ writer = wb.new_write()
+ commit = wb.new_commit()
+ try:
+ writer.write_arrow(current)
+ compact_messages = writer.prepare_commit()
+ assert len(compact_messages) == 1
+ message = compact_messages[0]
+ assert len(message.new_files) == 1
+ message.new_files = [message.new_files[0].assign_first_row_id(0)]
+ message.deleted_files.extend(file for split in splits for file in split.files)
+ # There is no public compaction API; publish the physical file
+ # replacement through the production committer as a COMPACT snapshot.
+ store_commit = commit.file_store_commit
+ try_commit = store_commit._try_commit
+ with patch.object(store_commit, "_try_commit",
+ side_effect=lambda commit_kind, **kwargs: try_commit("COMPACT", **kwargs)):
+ commit.commit(compact_messages)
+ compacted.append((table.snapshot_manager().get_latest_snapshot(), message.new_files[0].file_name))
+ finally:
+ writer.close()
+ commit.close()
+ return messages
+
+ def rollback_to(instant, from_snapshot):
+ assert table.snapshot_manager().get_latest_snapshot().id == from_snapshot
+ # Exercise real rollback if the regression returns: only the catalog
+ # endpoint is replaced, not conflict detection or the commit retry loop.
+ table.rollback_helper().clean_larger_than(
+ table.snapshot_manager().get_snapshot_by_id(instant.snapshot_id))
+
+ rollback = Mock()
+ rollback.rollback_to.side_effect = rollback_to
+ with patch.object(table.catalog_environment, "catalog_table_rollback", return_value=rollback), \
+ patch.object(GlobalIndexBuilder, "build", compact_after_build):
+ with pytest.raises(RuntimeError, match="Global index maintenance conflict"):
+ docs.maintain_indexes([spec], rebuild=rebuild)
+ rollback.rollback_to.assert_not_called()
+ snapshot, compacted_file = compacted[0]
+ latest = table.snapshot_manager().get_latest_snapshot()
+ assert (latest.id, latest.commit_kind) == (snapshot.id, "COMPACT")
+ assert compacted_file not in old_files
+ assert {file.file_name for split in table.new_read_builder().new_scan().plan().splits()
+ for file in split.files} == {compacted_file}
+ assert names(docs) == old_indexes
+ assert docs.scan().to_arrow().sort_by("id").to_pylist() == [{"id": 0}, {"id": 1}]
+ assert docs.maintain_indexes([spec], rebuild=rebuild) == 1
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot.id + 1
+
+
+def test_build_failure_keeps_indexes_and_cleans_completed_outputs(docs):
+ docs.maintain_indexes([SPEC])
+ old = names(docs)
+ original = GlobalIndexBuilder.build
+ outputs = []
+
+ def fail_second(builder):
+ if outputs:
+ raise RuntimeError("second index failed")
+ messages = original(builder)
+ outputs.extend(entry.index_file.file_name for message in messages for entry in message.index_adds)
+ return messages
+
+ with patch.object(GlobalIndexBuilder, "build", fail_second):
+ with pytest.raises(RuntimeError, match="second index failed"):
+ docs.maintain_indexes([SPEC, {"column": "id", "type": "btree"}], rebuild=True)
+ assert names(docs) == old
+ paths = docs.raw_table.path_factory().global_index_path_factory()
+ assert all(not docs.raw_table.file_io.exists(paths.to_path(name)) for name in outputs)
+
+
+@pytest.mark.parametrize("specs", [[], [{}], [SPEC, SPEC], [{"column": "missing", "type": "ivf-flat"}]])
+def test_validates_all_definitions_before_building(docs, specs):
+ with patch.object(GlobalIndexBuilder, "build", side_effect=AssertionError("premature build")):
+ with pytest.raises(ValueError):
+ docs.maintain_indexes(specs)
+
+
+def test_multiple_definitions_publish_once(docs):
+ before = docs.raw_table.snapshot_manager().get_latest_snapshot().id
+ assert docs.maintain_indexes([SPEC, {"column": "id", "type": "btree"}]) == 4
+ assert docs.raw_table.snapshot_manager().get_latest_snapshot().id == before + 1
+ assert {entry.index_file.index_type for entry in entries(docs)} == {"ivf-flat", "btree"}
+
+
+@pytest.mark.parametrize("rebuild", [False, True])
+def test_maintenance_keeps_python_planning_with_native_writes(docs, rebuild):
+ spec = {"column": "id", "type": "btree"}
+ docs.maintain_indexes([spec])
+ old = names(docs)
+ docs.add([{"id": 4, "pt": "a", "embedding": [4., 1.]}])
+ before = docs.raw_table.snapshot_manager().get_latest_snapshot().id
+ docs.raw_table = docs.raw_table.copy({
+ "write.native.enabled": "true", "deletion-vectors.enabled": "false"})
+ # Maintenance must use its pinned snapshot and Python rebuild ranges even
+ # when the table is eligible for the new native index-build entry point.
+ with patch("pypaimon.globalindex.native_index_build.native_method_available",
+ side_effect=AssertionError("maintenance delegated to native index planning")):
+ assert docs.maintain_indexes([spec], rebuild=rebuild) == (3 if rebuild else 1)
Review Comment:
[P2] Validate rebuild coverage instead of assuming three files
This new assertion fails on the exact-head Native CI job 114681964043 (`assert 2 == 3`). Native writers emit partition messages from a HashMap, so the initial row IDs may be b=0..1 and a=2..3. The later append to a receives row ID 4, and the Python BTree builder correctly merges a=2..4 into one contiguous range: two replacement files cover all five rows. I reproduced that partition order locally and verified all five IDs through actual indexed filters, old index replacement, and a single commit. Please assert those behaviors, or derive the count from the actual ranges, so valid native partition ordering does not fail CI.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Add atomic global index maintenance and replacement [paimon]
Posted by "TheR1sing3un (via GitHub)" <gi...@apache.org>.
TheR1sing3un commented on PR #10431:
URL: https://github.com/apache/paimon/pull/10431#issuecomment-6160191254
> I don't want to add this jarring API to Python. Is there a reference for it? How is it designed in Lance?
I agree. Then I'll temporarily suspend this pr and reopen it when needed later~
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Add atomic global index maintenance and replacement [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10431:
URL: https://github.com/apache/paimon/pull/10431#issuecomment-6080839855
[P2] Reject snapshot changes without rolling back concurrent compaction
The new strict snapshot check in `ConflictDetection.check_conflicts` returns a normal commit conflict. Maintenance sets `check_from_snapshot` on every message, which also sets `allow_rollback=True` in `FileStoreCommit.commit`. When a rollback-capable catalog has published a concurrent `COMPACT`, `_try_commit_once` consequently calls `CommitRollback.try_to_rollback`, rolls the table back to the preceding snapshot, and retries the maintenance commit. It can then publish successfully instead of rejecting the stale plan.
This contradicts the documented strict guard and lets a scheduled index maintenance pass undo a completed compaction. Please make this snapshot-change conflict non-rollbackable, or disable compaction rollback for maintenance/index-only messages. The existing concurrent-append test uses a filesystem catalog without catalog rollback and does not exercise this branch.
Validation: 90 targeted tests and 22 subtests passed locally, plus project Flake8. I also reproduced the rollback using real Parquet data, a real BTree rebuild and the production committer; only the catalog rollback endpoint was injected to delegate to local rollback. Planned snapshot 2 was followed by a physical-file-replacing COMPACT 3. Current code called rollback 3→2 and published maintenance as APPEND 3, restoring the original data file. Using the pre-PR conflict-check implementation instead published APPEND 4 with zero rollback calls and preserved the compacted file. Both variants retained the two logical rows; this finding does not claim data loss.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Add atomic global index maintenance and replacement [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10431:
URL: https://github.com/apache/paimon/pull/10431#discussion_r4244881041
##########
paimon-python/pypaimon/write/commit/conflict_detection.py:
##########
@@ -282,6 +282,12 @@ def check_conflicts(
if conflict is not None:
return conflict
+ if (not delta_entries and self._row_id_check_from_snapshot is not None
+ and any(entry.index_file.global_index_meta is not None for entry in (delta_index_entries or []))
+ and (latest_snapshot is None or latest_snapshot.id != self._row_id_check_from_snapshot)):
Review Comment:
[P1] Validate the captured snapshot identity before publishing maintenance
Snapshot IDs can be reused after public rollback, so comparing only the ID does not enforce the new strict snapshot guard. I reproduced this with a data-evolution table: append 0 (snapshot 1), append 10 (snapshot 2), start BTree maintenance, and after its build completes call rollback_to(1) then append 20. The replacement snapshot has ID 2 and the same schema/next_row_id, but a different UUID and data. Maintenance accepts it and publishes its old index for [0,10] as snapshot 3. A normal filter id=20 returns [{id:20}], while the same filter with the published global index returns no rows.
Row-range validation accepts the reused row IDs, and commit CAS uses the replacement snapshot's UUID read during publication, so neither protects the original plan. Please carry the captured snapshot identity into the maintenance commit and compare it on every publication attempt, using UUID or an equivalent immutable identity in addition to ID. Add a rollback-and-reappend regression that rejects this stale plan and preserves correct filtering.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org