[python] Fix index_manifest not inherited from previous snapshot with append commit (#7662)

Author: XiaoHongbo-Hope

paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py

@@ -393,6 +393,7 @@ def test_read_btree_index_table(self):
         self._test_read_btree_index_generic("test_btree_index_bigint", 2000, pa.int64())
         self._test_read_btree_index_large()
         self._test_read_btree_index_null()
+        self._test_index_manifest_inherited_after_write()
 
     def _test_read_btree_index_generic(self, table_name: str, k, k_type):
         table = self.catalog.get_table('default.' + table_name)
@@ -491,6 +492,29 @@ def _test_read_btree_index_null(self):
         })
         self.assertEqual(expected, actual)
 
+    def _test_index_manifest_inherited_after_write(self):
+        table = self.catalog.get_table('default.test_btree_index_string')
+
+        snapshot_before = table.snapshot_manager().get_latest_snapshot()
+        self.assertIsNotNone(snapshot_before.index_manifest,
+                             "Index manifest should exist before Python write")
+
+        write_builder = table.new_batch_write_builder()
+        write = write_builder.new_write()
+        commit = write_builder.new_commit()
+        data = pa.table({'k': ['k4'], 'v': ['v4']})
+        write.write_arrow(data)
+        commit.commit(write.prepare_commit())
+        write.close()
+        commit.close()
+
+        snapshot_after = table.snapshot_manager().get_latest_snapshot()
+        self.assertGreater(snapshot_after.id, snapshot_before.id)
+        self.assertIsNotNone(
+            snapshot_after.index_manifest,
+            "index_manifest lost after Python data write - indexes become invisible"
+        )
+
     @parameterized.expand([('json',), ('csv',)])
     def test_read_compressed_text_append_table(self, file_format):
         table = self.catalog.get_table(

paimon-python/pypaimon/tests/file_store_commit_test.py

@@ -18,7 +18,7 @@
 
 import unittest
 from datetime import datetime
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.manifest_entry import ManifestEntry
@@ -404,6 +404,49 @@ def test_generate_partition_statistics_empty_commit_messages(
         # Verify results
         self.assertEqual(len(statistics), 0)
 
+    def test_append_commit_inherits_index_manifest(
+            self, mock_manifest_list_manager, mock_manifest_file_manager, mock_snapshot_manager):
+        file_store_commit = self._create_file_store_commit()
+
+        self.mock_table.identifier = 'default.test_table'
+        self.mock_table.table_schema = Mock()
+        self.mock_table.table_schema.id = 7
+        self.mock_table.options.row_tracking_enabled.return_value = False
+
+        snapshot_commit = MagicMock()
+        snapshot_commit.__enter__.return_value = snapshot_commit
+        snapshot_commit.__exit__.return_value = False
+        snapshot_commit.commit.return_value = True
+        file_store_commit.snapshot_commit = snapshot_commit
+
+        file_store_commit._write_manifest_file = Mock(return_value=Mock())
+        file_store_commit._generate_partition_statistics = Mock(return_value=[])
+        file_store_commit.manifest_list_manager.read_all.return_value = []
+
+        latest_snapshot = Mock()
+        latest_snapshot.id = 3
+        latest_snapshot.total_record_count = 10
+        latest_snapshot.index_manifest = "index-manifest-existing"
+
+        commit_entry = Mock()
+        commit_entry.kind = 0
+        commit_entry.file = Mock()
+        commit_entry.file.row_count = 2
+
+        result = file_store_commit._try_commit_once(
+            retry_result=None,
+            commit_kind="APPEND",
+            commit_entries=[commit_entry],
+            commit_identifier=11,
+            latest_snapshot=latest_snapshot
+        )
+
+        self.assertTrue(result.is_success())
+        self.assertEqual(
+            "index-manifest-existing",
+            snapshot_commit.commit.call_args[0][0].index_manifest
+        )
+
     def test_null_partition_value(
             self, mock_manifest_list_manager, mock_manifest_file_manager, mock_snapshot_manager):
         from pypaimon.data.timestamp import Timestamp

paimon-python/pypaimon/write/file_store_commit.py

@@ -381,6 +381,10 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str
                     delta_record_count -= entry.file.row_count
 
             total_record_count += delta_record_count
+            index_manifest = None
+            if latest_snapshot and commit_kind == "APPEND":
+                index_manifest = latest_snapshot.index_manifest
+
             snapshot_data = Snapshot(
                 version=3,
                 id=new_snapshot_id,
@@ -394,6 +398,7 @@ def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str
                 commit_kind=commit_kind,
                 time_millis=int(time.time() * 1000),
                 next_row_id=next_row_id,
+                index_manifest=index_manifest,
             )
             # Generate partition statistics for the commit
             statistics = self._generate_partition_statistics(commit_entries)