[python] Isolate layout-sensitive tests from automatic compaction

paimon-python/pypaimon/tests/data_file_creation_time_test.py

@@ -101,7 +101,7 @@ def test_create_preserves_an_explicit_creation_timestamp():
     schema = pa.schema([('id', pa.int32()), ('pt', pa.string())])
     catalog.create_table('default.t', Schema.from_pyarrow_schema(schema, partition_keys=['pt'], options={
         'write.native.enabled': 'false', 'read.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
-        'commit.native.enabled': sys.argv[2]}), False)
+        'commit.native.enabled': sys.argv[2], 'write-only': 'true'}), False)
     table = catalog.get_table('default.t')
     captured = []
     original = server._table_commit_handle

paimon-python/pypaimon/tests/native_nullable_primary_key_test.py

@@ -140,6 +140,7 @@ def test_nullable_primary_key_source_uses_sql_merge_equality(rest_catalog, nativ
     source = _table(rest_catalog[0], 'nullable_source', -1, engine)
     _write(source, [(None, None, 10), (None, 1, 11), (1, None, 12), (1, 1, 100)], False)
     _write(source, [(None, None, 20), (None, 1, 21), (1, None, 22), (1, 1, 200)], True)
+    source_snapshot_id = source.snapshot_manager().get_latest_snapshot().id
     target = _table(rest_catalog[0], 'target', -1, primary_key=False)
     _write(target, [(None, None, -1), (None, 1, -3), (1, None, -4), (1, 1, -2)], False)
     target = target.copy({'write.native.enabled': str(native).lower()})
@@ -166,7 +167,7 @@ def test_nullable_primary_key_source_uses_sql_merge_equality(rest_catalog, nativ
             (1, None, -4), (1, None, values[2]), (1, 1, values[3])]
     assert _read(source, True, all_files=engine == 'first-row') == [
         (None, None, values[0]), (None, 1, values[1]), (1, None, values[2]), (1, 1, values[3])]
-    assert source.snapshot_manager().get_latest_snapshot().id == 2
+    assert source.snapshot_manager().get_latest_snapshot().id == source_snapshot_id
 
 
 @pytest.mark.parametrize('bucket', [3, -1], ids=['fixed', 'dynamic'])

paimon-python/pypaimon/tests/native_shard_update_test.py

@@ -130,6 +130,8 @@ def test_native_shard_updates_use_the_captured_reader_order(native_rest_catalog,
 @pytest.mark.parametrize('count', [1, 3, 7])
 def test_native_shards_are_balanced_by_file_group(native_rest_catalog, partitions, count):
     table, schema = _table(native_rest_catalog, partitions=partitions)
+    # Keep five independent file groups for the shard distribution check.
+    table = table.copy({'write-only': 'true'})
     for i in range(5):
         _append(table, [dict(p=str(i % 2), id=i, value=i * 10)], schema)
     snapshot_id = table.snapshot_manager().get_latest_snapshot().id

paimon-python/pypaimon/tests/rest/rest_catalog_test.py

@@ -31,7 +31,7 @@ def test_catalog_rollback(self):
         """Test table rollback to snapshot and tag."""
         table_name = "default.table_for_rollback"
         pa_schema = pa.schema([('col1', pa.int32())])
-        schema = Schema.from_pyarrow_schema(pa_schema)
+        schema = Schema.from_pyarrow_schema(pa_schema, options={"write-only": "true"})
         self.rest_catalog.create_table(table_name, schema, False)
         table = self.rest_catalog.get_table(table_name)
         identifier = Identifier.from_string(table_name)
@@ -167,7 +167,7 @@ def test_table_rollback_to_snapshot(self):
         """Test table-level rollback_to_snapshot via FileStoreTable."""
         table_name = "default.table_level_rollback_snapshot"
         pa_schema = pa.schema([('col1', pa.int32())])
-        schema = Schema.from_pyarrow_schema(pa_schema)
+        schema = Schema.from_pyarrow_schema(pa_schema, options={"write-only": "true"})
         self.rest_catalog.create_table(table_name, schema, False)
         table = self.rest_catalog.get_table(table_name)
 
@@ -196,7 +196,7 @@ def test_table_rollback_to_tag(self):
         """Test table-level rollback_to_tag via FileStoreTable."""
         table_name = "default.table_level_rollback_tag"
         pa_schema = pa.schema([('col1', pa.int32())])
-        schema = Schema.from_pyarrow_schema(pa_schema)
+        schema = Schema.from_pyarrow_schema(pa_schema, options={"write-only": "true"})
         self.rest_catalog.create_table(table_name, schema, False)
         table = self.rest_catalog.get_table(table_name)
 
@@ -409,7 +409,7 @@ def test_catalog_load_snapshot_after_rollback(self):
         """Test load_snapshot returns correct snapshot after rollback."""
         table_name = "default.table_load_snapshot_after_rollback"
         pa_schema = pa.schema([('col1', pa.int32())])
-        schema = Schema.from_pyarrow_schema(pa_schema)
+        schema = Schema.from_pyarrow_schema(pa_schema, options={"write-only": "true"})
         self.rest_catalog.create_table(table_name, schema, False)
         table = self.rest_catalog.get_table(table_name)
         identifier = Identifier.from_string(table_name)