You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mr-brobot (via GitHub)" <gi...@apache.org> on 2023/07/08 23:33:55 UTC
[GitHub] [iceberg] mr-brobot opened a new pull request, #8015: Python: Support serverless environments
mr-brobot opened a new pull request, #8015:
URL: https://github.com/apache/iceberg/pull/8015
Closes #8011
There's a lot of context on the related issue. Open to any feedback, from directional to the nittiest of picks. Despite being thin wrappers for built-in classes, `pyiceberg.utils.concurrent` lacks tests. Happy to address once I have the nod from folks on this approach.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] corleyma commented on pull request #8015: Python: Support serverless environments
Posted by "corleyma (via GitHub)" <gi...@apache.org>.
corleyma commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1635238669
I think that's a great call! Multiprocessing is tricky to get right, would be good to know the juice is worth the squeeze.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1258298648
##########
python/pyiceberg/table/__init__.py:
##########
@@ -773,11 +779,11 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entries: List[ManifestEntry] = []
positional_delete_entries = SortedList(key=lambda entry: entry.data_sequence_number or INITIAL_SEQUENCE_NUMBER)
- with ThreadPool() as pool:
+ with DynamicExecutor() as executor:
Review Comment:
Sure! What were you thinking for the config key and accepted values? Currently, this uses `PYICEBERG_CONCURRENCY_MODE` where possible values are `None`, `thread`, or `process`.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1262601138
##########
python/pyiceberg/io/pyarrow.py:
##########
@@ -521,6 +521,10 @@ def _construct_fragment(fs: FileSystem, data_file: DataFile, file_format_kwargs:
return _get_file_format(data_file.file_format, **file_format_kwargs).make_fragment(path, fs)
+def _starmap_read_deletes(args: Tuple[FileSystem, DataFile]) -> Dict[str, pa.ChunkedArray]:
Review Comment:
For me, this passthrough function does not add a lot of value. WDYT of:
```python
def _read_deletes(args: Tuple[FileSystem, DataFile]) -> Dict[str, pa.ChunkedArray]:
fs, data_file = args
```
##########
python/pyiceberg/table/__init__.py:
##########
@@ -773,11 +779,11 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entries: List[ManifestEntry] = []
positional_delete_entries = SortedList(key=lambda entry: entry.data_sequence_number or INITIAL_SEQUENCE_NUMBER)
- with ThreadPool() as pool:
+ with DynamicExecutor() as executor:
Review Comment:
Yes, this works for me. Can you also add this to the docs under `python/mkdocs/`?
##########
python/pyiceberg/table/__init__.py:
##########
@@ -639,6 +639,12 @@ def __init__(
self.length = length or data_file.file_size_in_bytes
+def _starmap_open_manifest(
Review Comment:
Same here, maybe better to let `_open_manifest` accept a single tuple argument, and then we can directly expand those.
##########
python/pyiceberg/io/pyarrow.py:
##########
@@ -725,6 +729,22 @@ def primitive(self, primitive: pa.DataType) -> IcebergType:
raise TypeError(f"Unsupported type: {primitive}")
+def _starmap_task_to_table(
Review Comment:
Same here, maybe better to let `_task_to_table` accept a single tuple argument, and then we can directly expand those.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1257414856
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,153 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import os
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from typing import (
+ Any,
+ Generic,
+ Type,
+ TypeVar,
+ Union,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.typedef import RecursiveDict
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(concurrency_mode: Union[str, RecursiveDict, None], shm_avail: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if concurrency_mode == "process":
+ return ProcessPoolExecutor
+ if concurrency_mode == "thread":
+ return ThreadPoolExecutor
+ if concurrency_mode is None and shm_avail:
+ return ProcessPoolExecutor
+ if concurrency_mode is None and not shm_avail:
+ logger.debug("Falling back to thread pool executor")
+ return ThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {concurrency_mode}")
+
+
+def _get_managed_executor_class(concurrency_mode: Union[str, RecursiveDict, None], shm_avail: bool) -> Type[ManagedExecutor]:
+ """Returns the managed executor class for the given concurrency mode."""
+ if concurrency_mode == "process":
+ return ManagedProcessPoolExecutor
+ if concurrency_mode == "thread":
+ return ManagedThreadPoolExecutor
+ if concurrency_mode is None and shm_avail:
+ return ManagedProcessPoolExecutor
+ if concurrency_mode is None and not shm_avail:
+ logger.debug("Falling back to managed thread pool executor")
+ return ManagedThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {concurrency_mode}")
+
+
+_shm_avail = os.path.exists("/dev/shm")
Review Comment:
Oh right, thanks. I'll find a more direct way to test for multi-processing support. Something like "try to create a `multiprocessing.Value`, if it fails we fall back on threads". Will change in the morning.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1634687728
> Just out curiosity, where are you able to run multi-processing?
@Fokko Yes, though I was on Ubuntu [where the start method defaults to `fork`](https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods) so this issue wasn't present. Since Windows & macOS default to the `spawn` method, and that is [considerably slower due to re-initialization of a new Python interpreter](https://peps.python.org/pep-0703/#multiprocessing), I think we should advise against multi-processing on these platforms in our documentation and perhaps log a warning.
In fact, in a previous version of this PR, I thought to fall back to multi-threading when the start mode was `spawn`:
```python
def _mp_pref() -> bool:
"""Returns whether multiprocessing is preferred."""
start_method = get_start_method()
if start_method != "fork":
logger.debug("Multi-processing not preferred in current runtime: start_method=%s", start_method)
return False
return start_method == "fork"
```
I'll get process spawning working and add some documentation. 😄
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1262641756
##########
python/pyiceberg/table/__init__.py:
##########
@@ -773,11 +779,11 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entries: List[ManifestEntry] = []
positional_delete_entries = SortedList(key=lambda entry: entry.data_sequence_number or INITIAL_SEQUENCE_NUMBER)
- with ThreadPool() as pool:
+ with DynamicExecutor() as executor:
Review Comment:
Apologies! I meant unset `PYICEBERG_CONCURRENCY_MODE`, which comes across as `None` in `pyiceberg.utils.concurrent._concurrency_mode`:
```python
def _concurrency_mode() -> ConcurrencyMode:
mode = Config().config.get("concurrency-mode")
if mode is None:
return "thread"
if mode in ("thread", "process"):
return mode # type: ignore
raise ValueError(f"Invalid concurrency mode: {mode}")
```
Will add some documentation!
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1635166281
@Fokko After some more testing in other platforms, I'm of the opinion that multi-processing support should be considered separately from simply making PyIceberg work in serverless environments.
1. Making PyIceberg work in serverless is simple. Replacing usage of `multiprocessing.pool.ThreadPool` and related synchronization concepts with `concurrent.futures.ThreadPoolExecutor` and the corresponding synchronization concepts in `threading`.
2. Making PyIceberg support multi-processing adds variability in behavior across platforms and would require more care when introducing future changes (e.g., ensuring that everything parallelized by an executor can be pickled).
I'm going to close this and reopen a separate PR focused on (1). For (2), I will create a separate issue where we can dedicate attention to testing and possibly some benchmarks that prove the ROI.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot closed pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot closed pull request #8015: Python: Support serverless environments
URL: https://github.com/apache/iceberg/pull/8015
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] corleyma commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "corleyma (via GitHub)" <gi...@apache.org>.
corleyma commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1257410026
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,153 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import os
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from typing import (
+ Any,
+ Generic,
+ Type,
+ TypeVar,
+ Union,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.typedef import RecursiveDict
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(concurrency_mode: Union[str, RecursiveDict, None], shm_avail: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if concurrency_mode == "process":
+ return ProcessPoolExecutor
+ if concurrency_mode == "thread":
+ return ThreadPoolExecutor
+ if concurrency_mode is None and shm_avail:
+ return ProcessPoolExecutor
+ if concurrency_mode is None and not shm_avail:
+ logger.debug("Falling back to thread pool executor")
+ return ThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {concurrency_mode}")
+
+
+def _get_managed_executor_class(concurrency_mode: Union[str, RecursiveDict, None], shm_avail: bool) -> Type[ManagedExecutor]:
+ """Returns the managed executor class for the given concurrency mode."""
+ if concurrency_mode == "process":
+ return ManagedProcessPoolExecutor
+ if concurrency_mode == "thread":
+ return ManagedThreadPoolExecutor
+ if concurrency_mode is None and shm_avail:
+ return ManagedProcessPoolExecutor
+ if concurrency_mode is None and not shm_avail:
+ logger.debug("Falling back to managed thread pool executor")
+ return ManagedThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {concurrency_mode}")
+
+
+_shm_avail = os.path.exists("/dev/shm")
Review Comment:
Pretty sure this doesn't work cross-platform.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1262616663
##########
python/pyiceberg/table/__init__.py:
##########
@@ -773,11 +779,11 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entries: List[ManifestEntry] = []
positional_delete_entries = SortedList(key=lambda entry: entry.data_sequence_number or INITIAL_SEQUENCE_NUMBER)
- with ThreadPool() as pool:
+ with DynamicExecutor() as executor:
Review Comment:
`None` doesn't seem to work on my end:
```
export PYICEBERG_CONCURRENCY_MODE=None
python3 /tmp/vo.py
Traceback (most recent call last):
File "/tmp/vo.py", line 1, in <module>
from pyiceberg.catalog import load_catalog
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/catalog/__init__.py", line 42, in <module>
from pyiceberg.serializers import ToOutputFile
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/serializers.py", line 26, in <module>
from pyiceberg.table.metadata import TableMetadata, TableMetadataUtil
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/table/__init__.py", line 72, in <module>
from pyiceberg.utils.concurrent import DynamicExecutor
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/utils/concurrent.py", line 146, in <module>
concurrency_mode = _concurrency_mode()
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/utils/concurrent.py", line 143, in _concurrency_mode
raise ValueError(f"Invalid concurrency mode: {mode}")
ValueError: Invalid concurrency mode: None
```
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1258249543
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
Review Comment:
Just to double-check if this is true. We're currently using a `ThreadPool`:
https://github.com/apache/iceberg/blob/e389e4d139624a49729379acd330dd9c96187b04/python/pyiceberg/io/pyarrow.py#L875
And:
https://github.com/apache/iceberg/blob/e389e4d139624a49729379acd330dd9c96187b04/python/pyiceberg/table/__init__.py#L776
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from multiprocessing import get_start_method
+from typing import (
+ Any,
+ Generic,
+ Optional,
+ Type,
+ TypeVar,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if mode == "process":
+ return ProcessPoolExecutor
+ if mode == "thread":
+ return ThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
+ logger.debug("Falling back to thread pool executor")
+ return ThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+
+def _get_managed_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[ManagedExecutor]:
Review Comment:
```suggestion
def _get_managed_executor_class(mode: Optional[Literal["thread", "process"]], mp_avail: bool, mp_pref: bool) -> Type[ManagedExecutor]:
```
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from multiprocessing import get_start_method
+from typing import (
+ Any,
+ Generic,
+ Optional,
+ Type,
+ TypeVar,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if mode == "process":
+ return ProcessPoolExecutor
+ if mode == "thread":
+ return ThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
+ logger.debug("Falling back to thread pool executor")
+ return ThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+
+def _get_managed_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[ManagedExecutor]:
+ """Returns the managed executor class for the given concurrency mode."""
+ if mode == "process":
+ return ManagedProcessPoolExecutor
+ if mode == "thread":
+ return ManagedThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ManagedProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
+ logger.debug("Falling back to managed thread pool executor")
+ return ManagedThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+
+def _concurrency_mode() -> Optional[str]:
+ mode = Config().config.get("concurrency-mode")
+
+ if mode not in ("thread", "process", None):
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+ return mode # type: ignore
+
+
+def _mp_avail() -> bool:
+ """Returns whether multiprocessing is available."""
+ try:
+ with ProcessPoolExecutor() as executor:
Review Comment:
I think this will already spin up new processes?
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from multiprocessing import get_start_method
+from typing import (
+ Any,
+ Generic,
+ Optional,
+ Type,
+ TypeVar,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if mode == "process":
+ return ProcessPoolExecutor
+ if mode == "thread":
+ return ThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
Review Comment:
Currently we use the ThreadPool, should we also use that as a default?
##########
python/pyiceberg/table/__init__.py:
##########
@@ -773,11 +779,11 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entries: List[ManifestEntry] = []
positional_delete_entries = SortedList(key=lambda entry: entry.data_sequence_number or INITIAL_SEQUENCE_NUMBER)
- with ThreadPool() as pool:
+ with DynamicExecutor() as executor:
Review Comment:
I think this is awesome. I have one more suggestion. What do you think of being able to set the executor through the config?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #8015: Python: Support serverless environments
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1634317651
Just out curiosity, where are you able to run multi-processing?
```
export PYICEBERG_CONCURRENCY_MODE=process
➜ docker-spark-iceberg git:(main) time python3 /tmp/vo.py
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 116, in spawn_main
exitcode = _main(fd, parent_sentinel)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 125, in _main
prepare(preparation_data)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 236, in prepare
_fixup_main_from_path(data['init_main_from_path'])
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 287, in _fixup_main_from_path
main_content = runpy.run_path(main_path,
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/runpy.py", line 268, in run_path
return _run_module_code(code, init_globals, run_name,
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/runpy.py", line 97, in _run_module_code
_run_code(code, mod_globals, init_globals,
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/runpy.py", line 87, in _run_code
exec(code, run_globals)
File "/tmp/vo.py", line 11, in <module>
df = table.scan().to_arrow()
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/table/__init__.py", line 825, in to_arrow
self.plan_files(),
File "/Users/fokkodriesprong/Desktop/iceberg/python/pyiceberg/table/__init__.py", line 784, in plan_files
*executor.map(
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/process.py", line 726, in map
results = super().map(partial(_process_chunk, fn),
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/_base.py", line 597, in map
fs = [self.submit(fn, *args) for args in zip(*iterables)]
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/_base.py", line 597, in <listcomp>
fs = [self.submit(fn, *args) for args in zip(*iterables)]
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/process.py", line 697, in submit
self._adjust_process_count()
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/concurrent/futures/process.py", line 675, in _adjust_process_count
p.start()
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 121, in start
self._popen = self._Popen(self)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/context.py", line 284, in _Popen
return Popen(process_obj)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/popen_spawn_posix.py", line 32, in __init__
super().__init__(process_obj)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/popen_fork.py", line 19, in __init__
self._launch(process_obj)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/popen_spawn_posix.py", line 42, in _launch
prep_data = spawn.get_preparation_data(process_obj._name)
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 154, in get_preparation_data
_check_not_importing_main()
File "/Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/multiprocessing/spawn.py", line 134, in _check_not_importing_main
raise RuntimeError('''
RuntimeError:
An attempt has been made to start a new process before the
current process has finished its bootstrapping phase.
This probably means that you are not using fork to start your
child processes and you have forgotten to use the proper idiom
in the main module:
if __name__ == '__main__':
freeze_support()
...
The "freeze_support()" line can be omitted if the program
is not going to be frozen to produce an executable.
```
Where:
```python
from pyiceberg.catalog import load_catalog
catalog = load_catalog('local')
table = catalog.load_table('nyc.taxis')
expected_rows = 15885533
for _ in range(10):
df = table.scan().to_arrow()
assert len(df) == expected_rows, f"Got {len(df)} rows, instead of {expected_rows}"
```
And local points to my local [docker-spark-iceberg](https://github.com/tabular-io/docker-spark-iceberg/tree/main/spark) setup, so I don't have S3 latency.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1258289394
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from multiprocessing import get_start_method
+from typing import (
+ Any,
+ Generic,
+ Optional,
+ Type,
+ TypeVar,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if mode == "process":
+ return ProcessPoolExecutor
+ if mode == "thread":
+ return ThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
Review Comment:
I think that makes the most sense. Making `ProcessPoolExecutor` only available via explicit config is also simpler, since there are a bunch of environments where it's either not supported or very slow (e.g., "spawning" environments like Windows & macOS).
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on a diff in pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on code in PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#discussion_r1258373103
##########
python/pyiceberg/utils/concurrent.py:
##########
@@ -0,0 +1,184 @@
+# 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.
+# pylint: disable=redefined-outer-name,arguments-renamed,fixme
+"""Concurrency concepts that adapt to the shared memory support in the current runtime.
+
+Performance-optimized concurrency in Python prefers `multiprocessing` to avoid the global
+interpreter lock. However, this requires shared memory provided via mount at `/dev/shm`. This
+is not provided in serverless runtimes. In environments where multiprocessing is not supported,
+we fall back to multithreading.
+"""
+import logging
+import multiprocessing
+import multiprocessing.managers
+import multiprocessing.synchronize
+import threading
+from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor
+from contextlib import AbstractContextManager
+from multiprocessing import get_start_method
+from typing import (
+ Any,
+ Generic,
+ Optional,
+ Type,
+ TypeVar,
+)
+
+from typing_extensions import Self
+
+from pyiceberg.utils.config import Config
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class Synchronized(Generic[T], AbstractContextManager): # type: ignore
+ """A context manager that provides concurrency-safe access to a value."""
+
+ value: T
+ lock: threading.Lock
+
+ def __init__(self, value: T, lock: threading.Lock):
+ super().__init__()
+ self.value = value
+ self.lock = lock
+
+ def __enter__(self) -> T:
+ """Acquires a lock, allowing access to the wrapped value."""
+ self.lock.acquire()
+ return self.value
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Releases the lock, allowing other threads to access the value."""
+ self.lock.release()
+
+
+class ManagedExecutor(Executor):
+ """An executor that provides synchronization."""
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ raise NotImplementedError
+
+
+class ManagedThreadPoolExecutor(ThreadPoolExecutor, ManagedExecutor):
+ """A thread pool executor that provides synchronization."""
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ super().__enter__()
+ return self
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = threading.Lock()
+ return Synchronized(value, lock)
+
+
+class ManagedProcessPoolExecutor(ProcessPoolExecutor, ManagedExecutor):
+ """A process pool executor provides synchronization."""
+
+ manager: multiprocessing.managers.SyncManager
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.manager = multiprocessing.Manager()
+
+ def __enter__(self) -> Self:
+ """Returns the executor itself as a context manager."""
+ self.manager.__enter__()
+ super().__enter__()
+ return self
+
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ """Exits the executor and the manager."""
+ super().__exit__(exc_type, exc_value, traceback)
+ self.manager.__exit__(exc_type, exc_value, traceback)
+
+ def synchronized(self, value: T) -> Synchronized[T]:
+ lock = self.manager.Lock()
+ return Synchronized(value, lock)
+
+
+def _get_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[Executor]:
+ """Returns the executor class for the given concurrency mode."""
+ if mode == "process":
+ return ProcessPoolExecutor
+ if mode == "thread":
+ return ThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
+ logger.debug("Falling back to thread pool executor")
+ return ThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+
+def _get_managed_executor_class(mode: Optional[str], mp_avail: bool, mp_pref: bool) -> Type[ManagedExecutor]:
+ """Returns the managed executor class for the given concurrency mode."""
+ if mode == "process":
+ return ManagedProcessPoolExecutor
+ if mode == "thread":
+ return ManagedThreadPoolExecutor
+ if mode is None and mp_avail and mp_pref:
+ return ManagedProcessPoolExecutor
+ if mode is None and (not mp_avail or not mp_pref):
+ logger.debug("Falling back to managed thread pool executor")
+ return ManagedThreadPoolExecutor
+
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+
+def _concurrency_mode() -> Optional[str]:
+ mode = Config().config.get("concurrency-mode")
+
+ if mode not in ("thread", "process", None):
+ raise ValueError(f"Invalid concurrency mode: {mode}")
+
+ return mode # type: ignore
+
+
+def _mp_avail() -> bool:
+ """Returns whether multiprocessing is available."""
+ try:
+ with ProcessPoolExecutor() as executor:
Review Comment:
Yeah, it was the only reliable way I could think of to check for multi-processing support across platforms... However, now that multi-threading is the default and multi-processing is only enabled via explicit `PYICEBERG_CONCURRENCY_MODE=process`, it should probably be the user's responsibility to decide if multi-processing is appropriate.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] koenvo commented on pull request #8015: Python: Support serverless environments
Posted by "koenvo (via GitHub)" <gi...@apache.org>.
koenvo commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1629685406
This looks very interesting and would solve some performance issues when switching to a process pool. It would be nice if a pool can be reused. Forking new processes for each new pool can be a costly operation
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] mr-brobot commented on pull request #8015: Python: Support serverless environments
Posted by "mr-brobot (via GitHub)" <gi...@apache.org>.
mr-brobot commented on PR #8015:
URL: https://github.com/apache/iceberg/pull/8015#issuecomment-1629782393
> It would be nice if a pool can be reused. Forking new processes for each new pool can be a costly operation
@koenvo Reusing the executor is a good idea. I can create a separate issue for that and work on it next.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org