refactor Config / lazy loading

Author: jx2lee

pyiceberg/catalog/__init__.py

@@ -25,6 +25,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from enum import Enum
+from functools import lru_cache
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -74,7 +75,11 @@
 
 logger = logging.getLogger(__name__)
 
-_ENV_CONFIG = Config()
+
+@lru_cache(maxsize=1)
+def _get_env_config() -> Config:
+    return Config.load()
+
 
 TOKEN = "token"
 TYPE = "type"
@@ -243,9 +248,9 @@ def load_catalog(name: str | None = None, **properties: str | None) -> Catalog:
             or if it could not determine the catalog based on the properties.
     """
     if name is None:
-        name = _ENV_CONFIG.get_default_catalog_name()
+        name = _get_env_config().get_default_catalog_name()
 
-    env = _ENV_CONFIG.get_catalog_config(name)
+    env = _get_env_config().get_catalog_config(name)
     conf: RecursiveDict = merge_config(env or {}, cast(RecursiveDict, properties))
 
     catalog_type: CatalogType | None
@@ -278,7 +283,7 @@ def load_catalog(name: str | None = None, **properties: str | None) -> Catalog:
 
 
 def list_catalogs() -> list[str]:
-    return _ENV_CONFIG.get_known_catalogs()
+    return _get_env_config().get_known_catalogs()
 
 
 def delete_files(io: FileIO, files_to_delete: set[str], file_type: str) -> None:
@@ -781,7 +786,7 @@ def _convert_schema_if_needed(
 
             from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow
 
-            downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+            downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
             if isinstance(schema, pa.Schema):
                 schema: Schema = visit_pyarrow(  # type: ignore
                     schema,

pyiceberg/catalog/bigquery_metastore.py

@@ -71,7 +71,7 @@ def __init__(self, name: str, **properties: str):
             raise ValueError(f"Missing property: {GCP_PROJECT_ID}")
 
         # BigQuery requires current-snapshot-id to be present for tables to be created.
-        if not Config().get_bool("legacy-current-snapshot-id"):
+        if not Config.load().get_bool("legacy-current-snapshot-id"):
             raise ValueError("legacy-current-snapshot-id must be enabled to work with BigQuery.")
 
         if credentials_file and credentials_info_str:

pyiceberg/io/pyarrow.py

@@ -1744,7 +1744,7 @@ def __init__(
         self._bound_row_filter = bind(table_metadata.schema(), row_filter, case_sensitive=case_sensitive)
         self._case_sensitive = case_sensitive
         self._limit = limit
-        self._downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE)
+        self._downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE)
 
     @property
     def _projected_field_ids(self) -> set[int]:
@@ -2685,7 +2685,7 @@ def write_parquet(task: WriteTask) -> DataFile:
         else:
             file_schema = table_schema
 
-        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         batches = [
             _to_requested_schema(
                 requested_schema=file_schema,
@@ -2892,7 +2892,7 @@ def _dataframe_to_data_files(
         default=TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT,
     )
     name_mapping = table_metadata.schema().name_mapping
-    downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+    downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
     task_schema = pyarrow_to_schema(
         df.schema,
         name_mapping=name_mapping,

pyiceberg/serializers.py

@@ -129,7 +129,7 @@ def table_metadata(metadata: TableMetadata, output_file: OutputFile, overwrite:
         """
         with output_file.create(overwrite=overwrite) as output_stream:
             # We need to serialize None values, in order to dump `None` current-snapshot-id as `-1`
-            exclude_none = False if Config().get_bool("legacy-current-snapshot-id") else True
+            exclude_none = False if Config.load().get_bool("legacy-current-snapshot-id") else True
 
             json_bytes = metadata.model_dump_json(exclude_none=exclude_none).encode(UTF8)
             json_bytes = Compressor.get_compressor(output_file.location).bytes_compressor()(json_bytes)

pyiceberg/table/__init__.py

@@ -469,7 +469,7 @@ def append(self, df: pa.Table, snapshot_properties: dict[str, str] = EMPTY_DICT,
         if not isinstance(df, pa.Table):
             raise ValueError(f"Expected PyArrow table, got: {df}")
 
-        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         _check_pyarrow_schema_compatible(
             self.table_metadata.schema(),
             provided_schema=df.schema,
@@ -523,7 +523,7 @@ def dynamic_partition_overwrite(
                     f"in the latest partition spec: {field}"
                 )
 
-        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         _check_pyarrow_schema_compatible(
             self.table_metadata.schema(),
             provided_schema=df.schema,
@@ -588,7 +588,7 @@ def overwrite(
         if not isinstance(df, pa.Table):
             raise ValueError(f"Expected PyArrow table, got: {df}")
 
-        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         _check_pyarrow_schema_compatible(
             self.table_metadata.schema(),
             provided_schema=df.schema,
@@ -787,7 +787,7 @@ def upsert(
 
         from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible
 
-        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
+        downcast_ns_timestamp_to_us = Config.load().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
         _check_pyarrow_schema_compatible(
             self.table_metadata.schema(),
             provided_schema=df.schema,

pyiceberg/table/metadata.py

@@ -328,7 +328,7 @@ def sort_order_by_id(self, sort_order_id: int) -> SortOrder | None:
 
     @field_serializer("current_snapshot_id")
     def serialize_current_snapshot_id(self, current_snapshot_id: int | None) -> int | None:
-        if current_snapshot_id is None and Config().get_bool("legacy-current-snapshot-id"):
+        if current_snapshot_id is None and Config.load().get_bool("legacy-current-snapshot-id"):
             return -1
         return current_snapshot_id
 

pyiceberg/utils/concurrent.py

@@ -29,7 +29,7 @@ class ExecutorFactory:
     @staticmethod
     def max_workers() -> int | None:
         """Return the max number of workers configured."""
-        return Config().get_int("max-workers")
+        return Config.load().get_int("max-workers")
 
     @staticmethod
     def get_or_create() -> Executor:

pyiceberg/utils/config.py

@@ -14,6 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
 import logging
 import os
 
@@ -59,10 +61,14 @@ def _lowercase_dictionary_keys(input_dict: RecursiveDict) -> RecursiveDict:
 class Config:
     config: RecursiveDict
 
-    def __init__(self) -> None:
-        config = self._from_configuration_files() or {}
-        config = merge_config(config, self._from_environment_variables(config))
-        self.config = FrozenDict(**config)
+    def __init__(self, config: RecursiveDict | None = None) -> None:
+        self.config = FrozenDict(**(config or {}))
+
+    @classmethod
+    def load(cls) -> Config:
+        config = cls._from_configuration_files() or {}
+        config = merge_config(config, cls._from_environment_variables(config))
+        return cls(config)
 
     @staticmethod
     def _from_configuration_files() -> RecursiveDict | None: