You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2026/01/08 04:11:33 UTC
(paimon) branch revert-6969-commitTest created (now 6e35a4eee0)
This is an automated email from the ASF dual-hosted git repository.
junhao pushed a change to branch revert-6969-commitTest
in repository https://gitbox.apache.org/repos/asf/paimon.git
at 6e35a4eee0 Revert "[python] Fix tryCommit failed. (#6969)"
This branch includes the following new commits:
new 6e35a4eee0 Revert "[python] Fix tryCommit failed. (#6969)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(paimon) 01/01: Revert "[python] Fix tryCommit failed. (#6969)"
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
junhao pushed a commit to branch revert-6969-commitTest
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 6e35a4eee0f9117c95d24f14dfaf3cc3d0b956bc
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Thu Jan 8 12:11:25 2026 +0800
Revert "[python] Fix tryCommit failed. (#6969)"
This reverts commit 2bacfdbf712827c1630b399db7b5f470211aed2d.
---
paimon-python/pypaimon/common/file_io.py | 18 +-
.../pypaimon/common/options/core_options.py | 48 ---
.../pypaimon/common/options/options_utils.py | 13 -
paimon-python/pypaimon/common/time_utils.py | 81 -----
.../pypaimon/snapshot/snapshot_manager.py | 47 +--
paimon-python/pypaimon/tests/blob_table_test.py | 135 --------
.../pypaimon/tests/reader_append_only_test.py | 106 -------
.../pypaimon/tests/reader_primary_key_test.py | 104 -------
.../pypaimon/tests/schema_evolution_read_test.py | 1 -
paimon-python/pypaimon/write/file_store_commit.py | 343 +++++----------------
10 files changed, 86 insertions(+), 810 deletions(-)
diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py
index 2ec1909306..497d711a49 100644
--- a/paimon-python/pypaimon/common/file_io.py
+++ b/paimon-python/pypaimon/common/file_io.py
@@ -18,15 +18,13 @@
import logging
import os
import subprocess
-import threading
-import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import splitport, urlparse
import pyarrow
from packaging.version import parse
-from pyarrow._fs import FileSystem, LocalFileSystem
+from pyarrow._fs import FileSystem
from pypaimon.common.options import Options
from pypaimon.common.options.config import OssOptions, S3Options
@@ -39,8 +37,6 @@ from pypaimon.write.blob_format_writer import BlobFormatWriter
class FileIO:
- rename_lock = threading.Lock()
-
def __init__(self, path: str, catalog_options: Options):
self.properties = catalog_options
self.logger = logging.getLogger(__name__)
@@ -255,15 +251,7 @@ class FileIO:
self.mkdirs(str(dst_parent))
src_str = self.to_filesystem_path(src)
- if isinstance(self.filesystem, LocalFileSystem):
- if self.exists(dst):
- return False
- with FileIO.rename_lock:
- if self.exists(dst):
- return False
- self.filesystem.move(src_str, dst_str)
- else:
- self.filesystem.move(src_str, dst_str)
+ self.filesystem.move(src_str, dst_str)
return True
except Exception as e:
self.logger.warning(f"Failed to rename {src} to {dst}: {e}")
@@ -315,7 +303,7 @@ class FileIO:
return input_stream.read().decode('utf-8')
def try_to_write_atomic(self, path: str, content: str) -> bool:
- temp_path = path + str(uuid.uuid4()) + ".tmp"
+ temp_path = path + ".tmp"
success = False
try:
self.write_file(temp_path, content, False)
diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py
index 49230240a7..4ab5a253d7 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -15,12 +15,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-import sys
from enum import Enum
from typing import Dict
-from datetime import timedelta
-
from pypaimon.common.memory_size import MemorySize
from pypaimon.common.options import Options
from pypaimon.common.options.config_options import ConfigOptions
@@ -242,34 +239,6 @@ class CoreOptions:
.with_description("The prefix for commit user.")
)
- COMMIT_MAX_RETRIES: ConfigOption[int] = (
- ConfigOptions.key("commit.max-retries")
- .int_type()
- .default_value(10)
- .with_description("Maximum number of retries for commit operations.")
- )
-
- COMMIT_TIMEOUT: ConfigOption[timedelta] = (
- ConfigOptions.key("commit.timeout")
- .duration_type()
- .no_default_value()
- .with_description("Timeout for commit operations (e.g., '10s', '5m'). If not set, effectively unlimited.")
- )
-
- COMMIT_MIN_RETRY_WAIT: ConfigOption[timedelta] = (
- ConfigOptions.key("commit.min-retry-wait")
- .duration_type()
- .default_value(timedelta(milliseconds=10))
- .with_description("Minimum wait time between commit retries (e.g., '10ms', '100ms').")
- )
-
- COMMIT_MAX_RETRY_WAIT: ConfigOption[timedelta] = (
- ConfigOptions.key("commit.max-retry-wait")
- .duration_type()
- .default_value(timedelta(seconds=10))
- .with_description("Maximum wait time between commit retries (e.g., '1s', '10s').")
- )
-
ROW_TRACKING_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("row-tracking.enabled")
.boolean_type()
@@ -421,20 +390,3 @@ class CoreOptions:
def data_file_external_paths_specific_fs(self, default=None):
return self.options.get(CoreOptions.DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS, default)
-
- def commit_max_retries(self) -> int:
- return self.options.get(CoreOptions.COMMIT_MAX_RETRIES)
-
- def commit_timeout(self) -> int:
- timeout = self.options.get(CoreOptions.COMMIT_TIMEOUT)
- if timeout is None:
- return sys.maxsize
- return int(timeout.total_seconds() * 1000)
-
- def commit_min_retry_wait(self) -> int:
- wait = self.options.get(CoreOptions.COMMIT_MIN_RETRY_WAIT)
- return int(wait.total_seconds() * 1000)
-
- def commit_max_retry_wait(self) -> int:
- wait = self.options.get(CoreOptions.COMMIT_MAX_RETRY_WAIT)
- return int(wait.total_seconds() * 1000)
diff --git a/paimon-python/pypaimon/common/options/options_utils.py b/paimon-python/pypaimon/common/options/options_utils.py
index 9938e87e74..f48f549df4 100644
--- a/paimon-python/pypaimon/common/options/options_utils.py
+++ b/paimon-python/pypaimon/common/options/options_utils.py
@@ -16,12 +16,10 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
-from datetime import timedelta
from enum import Enum
from typing import Any, Type
from pypaimon.common.memory_size import MemorySize
-from pypaimon.common.time_utils import parse_duration
class OptionsUtils:
@@ -65,8 +63,6 @@ class OptionsUtils:
return OptionsUtils.convert_to_double(value)
elif target_type == MemorySize:
return OptionsUtils.convert_to_memory_size(value)
- elif target_type == timedelta:
- return OptionsUtils.convert_to_duration(value)
else:
raise ValueError(f"Unsupported type: {target_type}")
@@ -129,15 +125,6 @@ class OptionsUtils:
return MemorySize.parse(value)
raise ValueError(f"Cannot convert {type(value)} to MemorySize")
- @staticmethod
- def convert_to_duration(value: Any) -> timedelta:
- if isinstance(value, timedelta):
- return value
- if isinstance(value, str):
- milliseconds = parse_duration(value)
- return timedelta(milliseconds=milliseconds)
- raise ValueError(f"Cannot convert {type(value)} to timedelta")
-
@staticmethod
def convert_to_enum(value: Any, enum_class: Type[Enum]) -> Enum:
diff --git a/paimon-python/pypaimon/common/time_utils.py b/paimon-python/pypaimon/common/time_utils.py
deleted file mode 100644
index 1a02bd4398..0000000000
--- a/paimon-python/pypaimon/common/time_utils.py
+++ /dev/null
@@ -1,81 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-
-def parse_duration(text: str) -> int:
- if text is None:
- raise ValueError("text cannot be None")
-
- trimmed = text.strip().lower()
- if not trimmed:
- raise ValueError("argument is an empty- or whitespace-only string")
-
- pos = 0
- while pos < len(trimmed) and trimmed[pos].isdigit():
- pos += 1
-
- number_str = trimmed[:pos]
- unit_str = trimmed[pos:].strip()
-
- if not number_str:
- raise ValueError("text does not start with a number")
-
- try:
- value = int(number_str)
- except ValueError:
- raise ValueError(
- f"The value '{number_str}' cannot be re represented as 64bit number (numeric overflow)."
- )
-
- if not unit_str:
- result_ms = value
- elif unit_str in ('ns', 'nano', 'nanosecond', 'nanoseconds'):
- result_ms = value / 1_000_000
- elif unit_str in ('µs', 'micro', 'microsecond', 'microseconds'):
- result_ms = value / 1_000
- elif unit_str in ('ms', 'milli', 'millisecond', 'milliseconds'):
- result_ms = value
- elif unit_str in ('s', 'sec', 'second', 'seconds'):
- result_ms = value * 1000
- elif unit_str in ('m', 'min', 'minute', 'minutes'):
- result_ms = value * 60 * 1000
- elif unit_str in ('h', 'hour', 'hours'):
- result_ms = value * 60 * 60 * 1000
- elif unit_str in ('d', 'day', 'days'):
- result_ms = value * 24 * 60 * 60 * 1000
- else:
- supported_units = (
- 'DAYS: (d | day | days), '
- 'HOURS: (h | hour | hours), '
- 'MINUTES: (m | min | minute | minutes), '
- 'SECONDS: (s | sec | second | seconds), '
- 'MILLISECONDS: (ms | milli | millisecond | milliseconds), '
- 'MICROSECONDS: (µs | micro | microsecond | microseconds), '
- 'NANOSECONDS: (ns | nano | nanosecond | nanoseconds)'
- )
- raise ValueError(
- f"Time interval unit label '{unit_str}' does not match any of the recognized units: "
- f"{supported_units}"
- )
-
- result_ms_int = int(round(result_ms))
-
- if result_ms_int < 0:
- raise ValueError(f"Duration cannot be negative: {text}")
-
- return result_ms_int
diff --git a/paimon-python/pypaimon/snapshot/snapshot_manager.py b/paimon-python/pypaimon/snapshot/snapshot_manager.py
index 8291d9cf2c..0d96563057 100644
--- a/paimon-python/pypaimon/snapshot/snapshot_manager.py
+++ b/paimon-python/pypaimon/snapshot/snapshot_manager.py
@@ -38,7 +38,7 @@ class SnapshotManager:
if not self.file_io.exists(self.latest_file):
return None
- latest_content = self.read_latest_file()
+ latest_content = self.file_io.read_file_utf8(self.latest_file)
latest_snapshot_id = int(latest_content.strip())
snapshot_file = f"{self.snapshot_dir}/snapshot-{latest_snapshot_id}"
@@ -48,51 +48,6 @@ class SnapshotManager:
snapshot_content = self.file_io.read_file_utf8(snapshot_file)
return JSON.from_json(snapshot_content, Snapshot)
- def read_latest_file(self, max_retries: int = 5):
- """
- Read the latest snapshot ID from LATEST file with retry mechanism.
- If file doesn't exist or is empty after retries, scan snapshot directory for max ID.
- """
- import re
- import time
-
- # Try to read LATEST file with retries
- for retry_count in range(max_retries):
- try:
- if self.file_io.exists(self.latest_file):
- content = self.file_io.read_file_utf8(self.latest_file)
- if content and content.strip():
- return content.strip()
-
- # File doesn't exist or is empty, wait a bit before retry
- if retry_count < max_retries - 1:
- time.sleep(0.001)
-
- except Exception:
- # On exception, wait and retry
- if retry_count < max_retries - 1:
- time.sleep(0.001)
-
- # List all files in snapshot directory
- file_infos = self.file_io.list_status(self.snapshot_dir)
-
- max_snapshot_id = None
- snapshot_pattern = re.compile(r'^snapshot-(\d+)$')
-
- for file_info in file_infos:
- # Get filename from path
- filename = file_info.path.split('/')[-1]
- match = snapshot_pattern.match(filename)
- if match:
- snapshot_id = int(match.group(1))
- if max_snapshot_id is None or snapshot_id > max_snapshot_id:
- max_snapshot_id = snapshot_id
-
- if not max_snapshot_id:
- raise RuntimeError(f"No snapshot content found in {self.snapshot_dir}")
-
- return str(max_snapshot_id)
-
def get_snapshot_path(self, snapshot_id: int) -> str:
"""
Get the path for a snapshot file.
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py
index f87f73ded7..de59d0398e 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -2567,141 +2567,6 @@ class DataBlobWriterTest(unittest.TestCase):
self.assertEqual(actual, expected)
- def test_concurrent_blob_writes_with_retry(self):
- """Test concurrent blob writes to verify retry mechanism works correctly."""
- import threading
- from pypaimon import Schema
- from pypaimon.snapshot.snapshot_manager import SnapshotManager
-
- # Run the test 10 times to verify stability
- iter_num = 2
- for test_iteration in range(iter_num):
- # Create a unique table for each iteration
- table_name = f'test_db.blob_concurrent_writes_{test_iteration}'
-
- # Create schema with blob column
- pa_schema = pa.schema([
- ('id', pa.int32()),
- ('thread_id', pa.int32()),
- ('metadata', pa.string()),
- ('blob_data', pa.large_binary()),
- ])
-
- schema = Schema.from_pyarrow_schema(
- pa_schema,
- options={
- 'row-tracking.enabled': 'true',
- 'data-evolution.enabled': 'true'
- }
- )
- self.catalog.create_table(table_name, schema, False)
- table = self.catalog.get_table(table_name)
-
- write_results = []
- write_errors = []
-
- # Create blob pattern for testing
- blob_size = 5 * 1024 # 5KB
- blob_pattern = b'BLOB_PATTERN_' + b'X' * 1024
- pattern_size = len(blob_pattern)
- repetitions = blob_size // pattern_size
- base_blob_data = blob_pattern * repetitions
-
- def write_blob_data(thread_id, start_id):
- """Write blob data in a separate thread."""
- try:
- threading.current_thread().name = f"Iter{test_iteration}-Thread-{thread_id}"
- write_builder = table.new_batch_write_builder()
- table_write = write_builder.new_write()
- table_commit = write_builder.new_commit()
-
- # Create unique blob data for this thread
- data = {
- 'id': list(range(start_id, start_id + 5)),
- 'thread_id': [thread_id] * 5,
- 'metadata': [f'thread{thread_id}_blob_{i}' for i in range(5)],
- 'blob_data': [i.to_bytes(2, byteorder='little') + base_blob_data for i in range(5)]
- }
- pa_table = pa.Table.from_pydict(data, schema=pa_schema)
-
- table_write.write_arrow(pa_table)
- commit_messages = table_write.prepare_commit()
-
- table_commit.commit(commit_messages)
- table_write.close()
- table_commit.close()
-
- write_results.append({
- 'thread_id': thread_id,
- 'start_id': start_id,
- 'success': True
- })
- except Exception as e:
- write_errors.append({
- 'thread_id': thread_id,
- 'error': str(e)
- })
-
- # Create and start multiple threads
- threads = []
- num_threads = 100
- for i in range(num_threads):
- thread = threading.Thread(
- target=write_blob_data,
- args=(i, i * 10)
- )
- threads.append(thread)
- thread.start()
-
- # Wait for all threads to complete
- for thread in threads:
- thread.join()
-
- # Verify all writes succeeded (retry mechanism should handle conflicts)
- self.assertEqual(num_threads, len(write_results),
- f"Iteration {test_iteration}: Expected {num_threads} successful writes, "
- f"got {len(write_results)}. Errors: {write_errors}")
- self.assertEqual(0, len(write_errors),
- f"Iteration {test_iteration}: Expected no errors, but got: {write_errors}")
-
- read_builder = table.new_read_builder()
- table_scan = read_builder.new_scan()
- table_read = read_builder.new_read()
- actual = table_read.to_arrow(table_scan.plan().splits()).sort_by('id')
-
- # Verify data rows
- self.assertEqual(num_threads * 5, actual.num_rows,
- f"Iteration {test_iteration}: Expected {num_threads * 5} rows")
-
- # Verify id column
- ids = actual.column('id').to_pylist()
- expected_ids = []
- for i in range(num_threads):
- expected_ids.extend(range(i * 10, i * 10 + 5))
- expected_ids.sort()
-
- self.assertEqual(ids, expected_ids,
- f"Iteration {test_iteration}: IDs mismatch")
-
- # Verify blob data integrity (spot check)
- blob_data_list = actual.column('blob_data').to_pylist()
- for i in range(0, len(blob_data_list), 100): # Check every 100th blob
- blob = blob_data_list[i]
- self.assertGreater(len(blob), 2, f"Blob {i} should have data")
- # Verify blob contains the pattern
- self.assertIn(b'BLOB_PATTERN_', blob, f"Blob {i} should contain pattern")
-
- # Verify snapshot count (should have num_threads snapshots)
- snapshot_manager = SnapshotManager(table)
- latest_snapshot = snapshot_manager.get_latest_snapshot()
- self.assertIsNotNone(latest_snapshot,
- f"Iteration {test_iteration}: Latest snapshot should not be None")
- self.assertEqual(latest_snapshot.id, num_threads,
- f"Iteration {test_iteration}: Expected snapshot ID {num_threads}, "
- f"got {latest_snapshot.id}")
-
- print(f"✓ Blob Table Iteration {test_iteration + 1}/{iter_num} completed successfully")
-
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py b/paimon-python/pypaimon/tests/reader_append_only_test.py
index b47f5d1f67..2661723919 100644
--- a/paimon-python/pypaimon/tests/reader_append_only_test.py
+++ b/paimon-python/pypaimon/tests/reader_append_only_test.py
@@ -17,7 +17,6 @@
################################################################################
import os
-import shutil
import tempfile
import time
import unittest
@@ -54,10 +53,6 @@ class AoReaderTest(unittest.TestCase):
'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2', 'p2'],
}, schema=cls.pa_schema)
- @classmethod
- def tearDownClass(cls):
- shutil.rmtree(cls.tempdir, ignore_errors=True)
-
def test_parquet_ao_reader(self):
schema = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=['dt'])
self.catalog.create_table('default.test_append_only_parquet', schema, False)
@@ -415,104 +410,3 @@ class AoReaderTest(unittest.TestCase):
table_read = read_builder.new_read()
splits = read_builder.new_scan().plan().splits()
return table_read.to_arrow(splits)
-
- def test_concurrent_writes_with_retry(self):
- """Test concurrent writes to verify retry mechanism works correctly."""
- import threading
-
- # Run the test 10 times to verify stability
- iter_num = 5
- for test_iteration in range(iter_num):
- # Create a unique table for each iteration
- table_name = f'default.test_concurrent_writes_{test_iteration}'
- schema = Schema.from_pyarrow_schema(self.pa_schema)
- self.catalog.create_table(table_name, schema, False)
- table = self.catalog.get_table(table_name)
-
- write_results = []
- write_errors = []
-
- def write_data(thread_id, start_user_id):
- """Write data in a separate thread."""
- try:
- threading.current_thread().name = f"Iter{test_iteration}-Thread-{thread_id}"
- write_builder = table.new_batch_write_builder()
- table_write = write_builder.new_write()
- table_commit = write_builder.new_commit()
-
- # Create unique data for this thread
- data = {
- 'user_id': list(range(start_user_id, start_user_id + 5)),
- 'item_id': [1000 + i for i in range(start_user_id, start_user_id + 5)],
- 'behavior': [f'thread{thread_id}_{i}' for i in range(5)],
- 'dt': ['p1' if i % 2 == 0 else 'p2' for i in range(5)],
- }
- pa_table = pa.Table.from_pydict(data, schema=self.pa_schema)
-
- table_write.write_arrow(pa_table)
- commit_messages = table_write.prepare_commit()
-
- table_commit.commit(commit_messages)
- table_write.close()
- table_commit.close()
-
- write_results.append({
- 'thread_id': thread_id,
- 'start_user_id': start_user_id,
- 'success': True
- })
- except Exception as e:
- write_errors.append({
- 'thread_id': thread_id,
- 'error': str(e)
- })
-
- # Create and start multiple threads
- threads = []
- num_threads = 100
- for i in range(num_threads):
- thread = threading.Thread(
- target=write_data,
- args=(i, i * 10)
- )
- threads.append(thread)
- thread.start()
-
- # Wait for all threads to complete
- for thread in threads:
- thread.join()
-
- # Verify all writes succeeded (retry mechanism should handle conflicts)
- self.assertEqual(num_threads, len(write_results),
- f"Iteration {test_iteration}: Expected {num_threads} successful writes, "
- f"got {len(write_results)}. Errors: {write_errors}")
- self.assertEqual(0, len(write_errors),
- f"Iteration {test_iteration}: Expected no errors, but got: {write_errors}")
-
- read_builder = table.new_read_builder()
- actual = self._read_test_table(read_builder).sort_by('user_id')
-
- # Verify data rows
- self.assertEqual(num_threads * 5, actual.num_rows,
- f"Iteration {test_iteration}: Expected {num_threads * 5} rows")
-
- # Verify user_id
- user_ids = actual.column('user_id').to_pylist()
- expected_user_ids = []
- for i in range(num_threads):
- expected_user_ids.extend(range(i * 10, i * 10 + 5))
- expected_user_ids.sort()
-
- self.assertEqual(user_ids, expected_user_ids,
- f"Iteration {test_iteration}: User IDs mismatch")
-
- # Verify snapshot count (should have num_threads snapshots)
- snapshot_manager = SnapshotManager(table)
- latest_snapshot = snapshot_manager.get_latest_snapshot()
- self.assertIsNotNone(latest_snapshot,
- f"Iteration {test_iteration}: Latest snapshot should not be None")
- self.assertEqual(latest_snapshot.id, num_threads,
- f"Iteration {test_iteration}: Expected snapshot ID {num_threads}, "
- f"got {latest_snapshot.id}")
-
- print(f"✓ Iteration {test_iteration + 1}/{iter_num} completed successfully")
diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py b/paimon-python/pypaimon/tests/reader_primary_key_test.py
index 731203385d..7077b2fd44 100644
--- a/paimon-python/pypaimon/tests/reader_primary_key_test.py
+++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py
@@ -422,107 +422,3 @@ class PkReaderTest(unittest.TestCase):
table_read = read_builder.new_read()
splits = read_builder.new_scan().plan().splits()
return table_read.to_arrow(splits)
-
- def test_concurrent_writes_with_retry(self):
- """Test concurrent writes to verify retry mechanism works correctly for PK tables."""
- import threading
-
- # Run the test 3 times to verify stability
- iter_num = 3
- for test_iteration in range(iter_num):
- # Create a unique table for each iteration
- table_name = f'default.test_pk_concurrent_writes_{test_iteration}'
- schema = Schema.from_pyarrow_schema(self.pa_schema,
- partition_keys=['dt'],
- primary_keys=['user_id', 'dt'],
- options={'bucket': '2'})
- self.catalog.create_table(table_name, schema, False)
- table = self.catalog.get_table(table_name)
-
- write_results = []
- write_errors = []
-
- def write_data(thread_id, start_user_id):
- """Write data in a separate thread."""
- try:
- threading.current_thread().name = f"Iter{test_iteration}-Thread-{thread_id}"
- write_builder = table.new_batch_write_builder()
- table_write = write_builder.new_write()
- table_commit = write_builder.new_commit()
-
- # Create unique data for this thread
- data = {
- 'user_id': list(range(start_user_id, start_user_id + 5)),
- 'item_id': [1000 + i for i in range(start_user_id, start_user_id + 5)],
- 'behavior': [f'thread{thread_id}_{i}' for i in range(5)],
- 'dt': ['p1' if i % 2 == 0 else 'p2' for i in range(5)],
- }
- pa_table = pa.Table.from_pydict(data, schema=self.pa_schema)
-
- table_write.write_arrow(pa_table)
- commit_messages = table_write.prepare_commit()
-
- table_commit.commit(commit_messages)
- table_write.close()
- table_commit.close()
-
- write_results.append({
- 'thread_id': thread_id,
- 'start_user_id': start_user_id,
- 'success': True
- })
- except Exception as e:
- write_errors.append({
- 'thread_id': thread_id,
- 'error': str(e)
- })
-
- # Create and start multiple threads
- threads = []
- num_threads = 100
- for i in range(num_threads):
- thread = threading.Thread(
- target=write_data,
- args=(i, i * 10)
- )
- threads.append(thread)
- thread.start()
-
- # Wait for all threads to complete
- for thread in threads:
- thread.join()
-
- # Verify all writes succeeded (retry mechanism should handle conflicts)
- self.assertEqual(num_threads, len(write_results),
- f"Iteration {test_iteration}: Expected {num_threads} successful writes, "
- f"got {len(write_results)}. Errors: {write_errors}")
- self.assertEqual(0, len(write_errors),
- f"Iteration {test_iteration}: Expected no errors, but got: {write_errors}")
-
- read_builder = table.new_read_builder()
- actual = self._read_test_table(read_builder).sort_by('user_id')
-
- # Verify data rows (PK table should have unique user_id+dt combinations)
- self.assertEqual(num_threads * 5, actual.num_rows,
- f"Iteration {test_iteration}: Expected {num_threads * 5} rows")
-
- # Verify user_id
- user_ids = actual.column('user_id').to_pylist()
- expected_user_ids = []
- for i in range(num_threads):
- expected_user_ids.extend(range(i * 10, i * 10 + 5))
- expected_user_ids.sort()
-
- self.assertEqual(user_ids, expected_user_ids,
- f"Iteration {test_iteration}: User IDs mismatch")
-
- # Verify snapshot count (should have num_threads snapshots)
- snapshot_manager = SnapshotManager(table)
- latest_snapshot = snapshot_manager.get_latest_snapshot()
- self.assertIsNotNone(latest_snapshot,
- f"Iteration {test_iteration}: Latest snapshot should not be None")
- self.assertEqual(latest_snapshot.id, num_threads,
- f"Iteration {test_iteration}: Expected snapshot ID {num_threads}, "
- f"got {latest_snapshot.id}")
-
- print(f"✓ PK Table Iteration {test_iteration + 1}/{iter_num} completed successfully")
diff --git a/paimon-python/pypaimon/tests/schema_evolution_read_test.py b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
index a67a927a5e..f5dafaae35 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
@@ -322,7 +322,6 @@ class SchemaEvolutionReadTest(unittest.TestCase):
# write schema-0 and schema-1 to table2
schema_manager = SchemaManager(table2.file_io, table2.table_path)
- schema_manager.file_io.delete_quietly(table2.table_path + "/schema/schema-0")
schema_manager.commit(TableSchema.from_schema(schema_id=0, schema=schema))
schema_manager.commit(TableSchema.from_schema(schema_id=1, schema=schema2))
diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py
index e55e25f7c8..a5b9fd9693 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -16,11 +16,9 @@
# limitations under the License.
################################################################################
-import logging
-import random
import time
import uuid
-from typing import List, Optional
+from typing import List
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
@@ -37,33 +35,6 @@ from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.write.commit_message import CommitMessage
-logger = logging.getLogger(__name__)
-
-
-class CommitResult:
- """Base class for commit results."""
-
- def is_success(self) -> bool:
- """Returns True if commit was successful."""
- raise NotImplementedError
-
-
-class SuccessResult(CommitResult):
- """Result indicating successful commit."""
-
- def is_success(self) -> bool:
- return True
-
-
-class RetryResult(CommitResult):
-
- def __init__(self, latest_snapshot, exception: Optional[Exception] = None):
- self.latest_snapshot = latest_snapshot
- self.exception = exception
-
- def is_success(self) -> bool:
- return False
-
class FileStoreCommit:
"""
@@ -87,11 +58,6 @@ class FileStoreCommit:
self.manifest_target_size = 8 * 1024 * 1024
self.manifest_merge_min_count = 30
- self.commit_max_retries = table.options.commit_max_retries()
- self.commit_timeout = table.options.commit_timeout()
- self.commit_min_retry_wait = table.options.commit_min_retry_wait()
- self.commit_max_retry_wait = table.options.commit_max_retry_wait()
-
def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
"""Commit the given commit messages in normal append mode."""
if not commit_messages:
@@ -133,81 +99,27 @@ class FileStoreCommit:
raise RuntimeError(f"Trying to overwrite partition {overwrite_partition}, but the changes "
f"in {msg.partition} does not belong to this partition")
- self._overwrite_partition_filter = partition_filter
- self._overwrite_commit_messages = commit_messages
+ commit_entries = []
+ current_entries = FullStartingScanner(self.table, partition_filter, None).plan_files()
+ for entry in current_entries:
+ entry.kind = 1
+ commit_entries.append(entry)
+ for msg in commit_messages:
+ partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
+ for file in msg.new_files:
+ commit_entries.append(ManifestEntry(
+ kind=0,
+ partition=partition,
+ bucket=msg.bucket,
+ total_buckets=self.table.total_buckets,
+ file=file
+ ))
- self._try_commit(
- commit_kind="OVERWRITE",
- commit_entries=None, # Will be generated in _try_commit based on latest snapshot
- commit_identifier=commit_identifier
- )
+ self._try_commit(commit_kind="OVERWRITE",
+ commit_entries=commit_entries,
+ commit_identifier=commit_identifier)
def _try_commit(self, commit_kind, commit_entries, commit_identifier):
- import threading
-
- retry_count = 0
- retry_result = None
- start_time_ms = int(time.time() * 1000)
- thread_id = threading.current_thread().name
- while True:
- latest_snapshot = self.snapshot_manager.get_latest_snapshot()
-
- if commit_kind == "OVERWRITE":
- commit_entries = self._generate_overwrite_entries()
-
- result = self._try_commit_once(
- retry_result=retry_result,
- commit_kind=commit_kind,
- commit_entries=commit_entries,
- commit_identifier=commit_identifier,
- latest_snapshot=latest_snapshot
- )
-
- if result.is_success():
- logger.warning(
- f"Thread {thread_id}: commit success {latest_snapshot.id + 1 if latest_snapshot else 1} "
- f"after {retry_count} retries"
- )
- break
-
- retry_result = result
-
- elapsed_ms = int(time.time() * 1000) - start_time_ms
- if elapsed_ms > self.commit_timeout or retry_count >= self.commit_max_retries:
- error_msg = (
- f"Commit failed {latest_snapshot.id + 1 if latest_snapshot else 1} "
- f"after {elapsed_ms} millis with {retry_count} retries, "
- f"there maybe exist commit conflicts between multiple jobs."
- )
- if retry_result.exception:
- raise RuntimeError(error_msg) from retry_result.exception
- else:
- raise RuntimeError(error_msg)
-
- self._commit_retry_wait(retry_count)
- retry_count += 1
-
- def _try_commit_once(self, retry_result: Optional[RetryResult], commit_kind: str,
- commit_entries: List[ManifestEntry], commit_identifier: int,
- latest_snapshot: Optional[Snapshot]) -> CommitResult:
- start_time_ms = int(time.time() * 1000)
-
- if retry_result is not None and latest_snapshot is not None:
- start_check_snapshot_id = 1 # Snapshot.FIRST_SNAPSHOT_ID
- if retry_result.latest_snapshot is not None:
- start_check_snapshot_id = retry_result.latest_snapshot.id + 1
-
- for snapshot_id in range(start_check_snapshot_id, latest_snapshot.id + 2):
- snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id)
- if (snapshot and snapshot.commit_user == self.commit_user and
- snapshot.commit_identifier == commit_identifier and
- snapshot.commit_kind == commit_kind):
- logger.info(
- f"Commit already completed (snapshot {snapshot_id}), "
- f"user: {self.commit_user}, identifier: {commit_identifier}"
- )
- return SuccessResult()
-
unique_id = uuid.uuid4()
base_manifest_list = f"manifest-list-{unique_id}-0"
delta_manifest_list = f"manifest-list-{unique_id}-1"
@@ -218,6 +130,7 @@ class FileStoreCommit:
deleted_file_count = 0
delta_record_count = 0
# process snapshot
+ latest_snapshot = self.snapshot_manager.get_latest_snapshot()
new_snapshot_id = latest_snapshot.id + 1 if latest_snapshot else 1
# Check if row tracking is enabled
@@ -230,7 +143,7 @@ class FileStoreCommit:
commit_entries = self._assign_snapshot_id(new_snapshot_id, commit_entries)
# Get the next row ID start from the latest snapshot
- first_row_id_start = self._get_next_row_id_start(latest_snapshot)
+ first_row_id_start = self._get_next_row_id_start()
# Assign row IDs to new files and get the next row ID for the snapshot
commit_entries, next_row_id = self._assign_row_tracking_meta(first_row_id_start, commit_entries)
@@ -242,164 +155,71 @@ class FileStoreCommit:
else:
deleted_file_count += 1
delta_record_count -= entry.file.row_count
-
- try:
- self.manifest_file_manager.write(new_manifest_file, commit_entries)
-
- # TODO: implement noConflictsOrFail logic
- partition_columns = list(zip(*(entry.partition.values for entry in commit_entries)))
- partition_min_stats = [min(col) for col in partition_columns]
- partition_max_stats = [max(col) for col in partition_columns]
- partition_null_counts = [sum(value == 0 for value in col) for col in partition_columns]
- if not all(count == 0 for count in partition_null_counts):
- raise RuntimeError("Partition value should not be null")
-
- manifest_file_path = f"{self.manifest_file_manager.manifest_path}/{new_manifest_file}"
- file_size = self.table.file_io.get_file_size(manifest_file_path)
-
- new_manifest_file_meta = ManifestFileMeta(
- file_name=new_manifest_file,
- file_size=file_size,
- num_added_files=added_file_count,
- num_deleted_files=deleted_file_count,
- partition_stats=SimpleStats(
- min_values=GenericRow(
- values=partition_min_stats,
- fields=self.table.partition_keys_fields
- ),
- max_values=GenericRow(
- values=partition_max_stats,
- fields=self.table.partition_keys_fields
- ),
- null_counts=partition_null_counts,
+ self.manifest_file_manager.write(new_manifest_file, commit_entries)
+ # TODO: implement noConflictsOrFail logic
+ partition_columns = list(zip(*(entry.partition.values for entry in commit_entries)))
+ partition_min_stats = [min(col) for col in partition_columns]
+ partition_max_stats = [max(col) for col in partition_columns]
+ partition_null_counts = [sum(value == 0 for value in col) for col in partition_columns]
+ if not all(count == 0 for count in partition_null_counts):
+ raise RuntimeError("Partition value should not be null")
+ manifest_file_path = f"{self.manifest_file_manager.manifest_path}/{new_manifest_file}"
+ new_manifest_list = ManifestFileMeta(
+ file_name=new_manifest_file,
+ file_size=self.table.file_io.get_file_size(manifest_file_path),
+ num_added_files=added_file_count,
+ num_deleted_files=deleted_file_count,
+ partition_stats=SimpleStats(
+ min_values=GenericRow(
+ values=partition_min_stats,
+ fields=self.table.partition_keys_fields
),
- schema_id=self.table.table_schema.id,
- )
-
- self.manifest_list_manager.write(delta_manifest_list, [new_manifest_file_meta])
-
- # process existing_manifest
- total_record_count = 0
- if latest_snapshot:
- existing_manifest_files = self.manifest_list_manager.read_all(latest_snapshot)
- previous_record_count = latest_snapshot.total_record_count
- if previous_record_count:
- total_record_count += previous_record_count
- else:
- existing_manifest_files = []
-
- self.manifest_list_manager.write(base_manifest_list, existing_manifest_files)
- total_record_count += delta_record_count
- snapshot_data = Snapshot(
- version=3,
- id=new_snapshot_id,
- schema_id=self.table.table_schema.id,
- base_manifest_list=base_manifest_list,
- delta_manifest_list=delta_manifest_list,
- total_record_count=total_record_count,
- delta_record_count=delta_record_count,
- commit_user=self.commit_user,
- commit_identifier=commit_identifier,
- commit_kind=commit_kind,
- time_millis=int(time.time() * 1000),
- next_row_id=next_row_id,
- )
- # Generate partition statistics for the commit
- statistics = self._generate_partition_statistics(commit_entries)
- except Exception as e:
- self._cleanup_preparation_failure(new_manifest_file, delta_manifest_list,
- base_manifest_list)
- logger.warning(f"Exception occurs when preparing snapshot: {e}", exc_info=True)
- raise RuntimeError(f"Failed to prepare snapshot: {e}")
-
- # Use SnapshotCommit for atomic commit
- try:
- with self.snapshot_commit:
- success = self.snapshot_commit.commit(snapshot_data, self.table.current_branch(), statistics)
- if not success:
- # Commit failed, clean up temporary files and retry
- commit_time_sec = (int(time.time() * 1000) - start_time_ms) / 1000
- logger.warning(
- f"Atomic commit failed for snapshot #{new_snapshot_id} "
- f"by user {self.commit_user} "
- f"with identifier {commit_identifier} and kind {commit_kind} after {commit_time_sec}s. "
- f"Clean up and try again."
- )
- self._cleanup_preparation_failure(new_manifest_file, delta_manifest_list,
- base_manifest_list)
- return RetryResult(latest_snapshot, None)
- except Exception as e:
- # Commit exception, not sure about the situation and should not clean up the files
- logger.warning("Retry commit for exception")
- return RetryResult(latest_snapshot, e)
-
- logger.warning(
- f"Successfully commit snapshot {new_snapshot_id} to table {self.table.identifier} "
- f"for snapshot-{new_snapshot_id} by user {self.commit_user} "
- + f"with identifier {commit_identifier} and kind {commit_kind}."
+ max_values=GenericRow(
+ values=partition_max_stats,
+ fields=self.table.partition_keys_fields
+ ),
+ null_counts=partition_null_counts,
+ ),
+ schema_id=self.table.table_schema.id,
)
- return SuccessResult()
+ self.manifest_list_manager.write(delta_manifest_list, [new_manifest_list])
+
+ # process existing_manifest
+ total_record_count = 0
+ if latest_snapshot:
+ existing_manifest_files = self.manifest_list_manager.read_all(latest_snapshot)
+ previous_record_count = latest_snapshot.total_record_count
+ if previous_record_count:
+ total_record_count += previous_record_count
+ else:
+ existing_manifest_files = []
+ self.manifest_list_manager.write(base_manifest_list, existing_manifest_files)
- def _generate_overwrite_entries(self):
- """Generate commit entries for OVERWRITE mode based on latest snapshot."""
- entries = []
- current_entries = FullStartingScanner(self.table, self._overwrite_partition_filter, None).plan_files()
- for entry in current_entries:
- entry.kind = 1 # DELETE
- entries.append(entry)
- for msg in self._overwrite_commit_messages:
- partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
- for file in msg.new_files:
- entries.append(ManifestEntry(
- kind=0, # ADD
- partition=partition,
- bucket=msg.bucket,
- total_buckets=self.table.total_buckets,
- file=file
- ))
- return entries
-
- def _commit_retry_wait(self, retry_count: int):
- import threading
- thread_id = threading.get_ident()
-
- retry_wait_ms = min(
- self.commit_min_retry_wait * (2 ** retry_count),
- self.commit_max_retry_wait
+ # process snapshot
+ total_record_count += delta_record_count
+ snapshot_data = Snapshot(
+ version=3,
+ id=new_snapshot_id,
+ schema_id=self.table.table_schema.id,
+ base_manifest_list=base_manifest_list,
+ delta_manifest_list=delta_manifest_list,
+ total_record_count=total_record_count,
+ delta_record_count=delta_record_count,
+ commit_user=self.commit_user,
+ commit_identifier=commit_identifier,
+ commit_kind=commit_kind,
+ time_millis=int(time.time() * 1000),
+ next_row_id=next_row_id,
)
- jitter_ms = random.randint(0, max(1, int(retry_wait_ms * 0.2)))
- total_wait_ms = retry_wait_ms + jitter_ms
+ # Generate partition statistics for the commit
+ statistics = self._generate_partition_statistics(commit_entries)
- logger.debug(
- f"Thread {thread_id}: Waiting {total_wait_ms}ms before retry (base: {retry_wait_ms}ms, "
- f"jitter: {jitter_ms}ms)"
- )
- time.sleep(total_wait_ms / 1000.0)
-
- def _cleanup_preparation_failure(self, manifest_file: Optional[str],
- delta_manifest_list: Optional[str],
- base_manifest_list: Optional[str]):
- try:
- manifest_path = self.manifest_list_manager.manifest_path
-
- if delta_manifest_list:
- manifest_files = self.manifest_list_manager.read(delta_manifest_list)
- for manifest_meta in manifest_files:
- manifest_file_path = f"{self.manifest_file_manager.manifest_path}/{manifest_meta.file_name}"
- self.table.file_io.delete_quietly(manifest_file_path)
- delta_path = f"{manifest_path}/{delta_manifest_list}"
- self.table.file_io.delete_quietly(delta_path)
-
- if base_manifest_list:
- base_path = f"{manifest_path}/{base_manifest_list}"
- self.table.file_io.delete_quietly(base_path)
-
- if manifest_file:
- manifest_file_path = f"{self.manifest_file_manager.manifest_path}/{manifest_file}"
- self.table.file_io.delete_quietly(manifest_file_path)
- except Exception as e:
- logger.warning(f"Failed to clean up temporary files during preparation failure: {e}", exc_info=True)
+ # Use SnapshotCommit for atomic commit
+ with self.snapshot_commit:
+ success = self.snapshot_commit.commit(snapshot_data, self.table.current_branch(), statistics)
+ if not success:
+ raise RuntimeError(f"Failed to commit snapshot {new_snapshot_id}")
def abort(self, commit_messages: List[CommitMessage]):
"""Abort commit and delete files. Uses external_path if available to ensure proper scheme handling."""
@@ -512,8 +332,9 @@ class FileStoreCommit:
"""Assign snapshot ID to all commit entries."""
return [entry.assign_sequence_number(snapshot_id, snapshot_id) for entry in commit_entries]
- def _get_next_row_id_start(self, latest_snapshot) -> int:
+ def _get_next_row_id_start(self) -> int:
"""Get the next row ID start from the latest snapshot."""
+ latest_snapshot = self.snapshot_manager.get_latest_snapshot()
if latest_snapshot and hasattr(latest_snapshot, 'next_row_id') and latest_snapshot.next_row_id is not None:
return latest_snapshot.next_row_id
return 0