Fix issue 2530

pyiceberg/catalog/bigquery_metastore.py

@@ -102,6 +102,16 @@ def __init__(self, name: str, **properties: str):
  self.location = location
  self.project_id = project_id
 
+ @override
+ def close(self) -> None:
+ """Close the catalog and release BigQuery client resources.
+
+ This method closes the BigQuery client connection. This is crucial for proper
+ resource cleanup and must be called in test fixture teardowns to prevent
+ Python 3.13+ ResourceWarnings about unclosed connections.
+ """
+ self.client.close()
+
  @override
  def create_table(
  self,

pyiceberg/catalog/sql.py

@@ -778,9 +778,25 @@ def close(self) -> None:
  """Close the catalog and release database connections.
 
  This method closes the SQLAlchemy engine and disposes of all connection pools.
- This ensures that any cached connections are properly closed, which is especially
- important for blobfuse scenarios where file handles need to be closed for
- data to be flushed to persistent storage.
+ This is crucial for proper resource cleanup and must be called in test fixture
+ teardowns to prevent Python 3.13+ ResourceWarnings about unclosed connections.
+
+ The connection pool lifecycle is managed by SQLAlchemy's engine. When dispose()
+ is called, it closes all idle connections and invalidates active ones. This is
+ especially important in test environments where fixtures create catalogs that
+ otherwise may not be explicitly closed, and in production scenarios with
+ blobfuse where file handles need to be closed for data to be flushed to
+ persistent storage.
+
+ For Python 3.14+, explicit garbage collection is needed to ensure all
+ sqlite3 connections are immediately released, as deferred GC can cause
+ ResourceWarnings during test teardown.
+
+ See: Issue #2530 (https://github.com/apache/iceberg-python/issues/2530)
  """
  if hasattr(self, "engine"):
  self.engine.dispose()
+ # Force immediate garbage collection to release sqlite3 connections on Python 3.14+
+ import gc
+
+ gc.collect()

pyiceberg/cli/console.py

@@ -117,6 +117,8 @@ def run(
  )
  ctx.exit(1)
 
+ ctx.call_on_close(lambda: ctx.obj["catalog"].close())
+
 
 def _catalog_and_output(ctx: Context) -> tuple[Catalog, Output]:
  """Small helper to set the types."""

pyiceberg/table/__init__.py

@@ -2471,8 +2471,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
  options=self.options,
  ).plan_files(
  manifests=manifests,
- manifest_entry_filter=lambda manifest_entry: manifest_entry.snapshot_id in append_snapshot_ids
- and manifest_entry.status == ManifestEntryStatus.ADDED,
+ manifest_entry_filter=lambda manifest_entry: (
+ manifest_entry.snapshot_id in append_snapshot_ids and manifest_entry.status == ManifestEntryStatus.ADDED
+ ),
  )
 
  def to_arrow(self) -> pa.Table:

pyproject.toml

@@ -124,6 +124,8 @@ dev = [
  "papermill>=2.6.0",
  "nbformat>=5.10.0",
  "ipykernel>=6.29.0",
+ "ruff>=0.15.22",
+ "pre-commit>=4.6.0",
 ]
 # for mkdocs
 docs = [
@@ -171,9 +173,10 @@ filterwarnings = [
  "error",
  # Ignore Python version deprecation warning from google.api_core while we still support 3.10
  "ignore:You are using a Python version.*which Google will stop supporting:FutureWarning:google.api_core",
- # Python 3.13 sqlite3 module ResourceWarnings for unclosed database connections
- "ignore:unclosed database in <sqlite3.Connection object*:ResourceWarning",
- # Ignore Ray subprocess cleanup warnings
+ # Ray subprocess cleanup warnings are suppressed because ray is not installed in local test environments
+ # and requires a CI environment with proper isolation to test subprocess lifecycle and cleanup behavior.
+ # See issue #2530: the SQLAlchemy connection pool warning (sqlite) has been fixed by properly closing
+ # connections in test fixtures, but ray subprocess warnings still require CI-level testing.
  "ignore:unclosed file:ResourceWarning",
  "ignore:subprocess.*is still running:ResourceWarning",
  # Ignore google-crc32c C extension missing warning (common on Python 3.14+)

tests/catalog/test_base.py

@@ -37,7 +37,11 @@ def catalog(tmp_path: PosixPath) -> Generator[Catalog, None, None]:
 
 
 def test_load_catalog_in_memory() -> None:
- assert load_catalog("catalog", type="in-memory")
+ catalog = load_catalog("catalog", type="in-memory")
+ try:
+ assert catalog
+ finally:
+ catalog.close()
 
 
 def test_load_catalog_impl_not_full_path() -> None:

tests/catalog/test_bigquery_metastore.py

@@ -67,9 +67,12 @@ def test_create_table_with_database_location(
  test_catalog = BigQueryMetastoreCatalog(
  catalog_name, **{"gcp.bigquery.project-id": "alexstephen-test-1", "warehouse": "gs://alexstephen-test-bq-bucket/"}
  )
- test_catalog.create_namespace(namespace=gcp_dataset_name)
- table = test_catalog.create_table(identifier, table_schema_nested)
- assert table.name() == identifier
+ try:
+ test_catalog.create_namespace(namespace=gcp_dataset_name)
+ table = test_catalog.create_table(identifier, table_schema_nested)
+ assert table.name() == identifier
+ finally:
+ test_catalog.close()
 
 
 def test_drop_table_with_database_location(
@@ -93,19 +96,22 @@ def test_drop_table_with_database_location(
  test_catalog = BigQueryMetastoreCatalog(
  catalog_name, **{"gcp.bigquery.project-id": "alexstephen-test-1", "warehouse": "gs://alexstephen-test-bq-bucket/"}
  )
- test_catalog.create_namespace(namespace=gcp_dataset_name)
- test_catalog.create_table(identifier, table_schema_nested)
- test_catalog.drop_table(identifier)
+ try:
+ test_catalog.create_namespace(namespace=gcp_dataset_name)
+ test_catalog.create_table(identifier, table_schema_nested)
+ test_catalog.drop_table(identifier)
 
- client_mock.get_table.side_effect = NotFound("Table Not Found")
- mocker.patch("pyiceberg.catalog.bigquery_metastore.Client", return_value=client_mock)
+client_mock.get_table.side_effect = NotFound("Table Not Found")
+mocker.patch("pyiceberg.catalog.bigquery_metastore.Client", return_value=client_mock)
 
- # Expect that the table no longer exists.
- try:
- test_catalog.load_table(identifier)
- raise AssertionError()
- except NoSuchTableError:
- assert True
+ # Expect that the table no longer exists.
+ try:
+ test_catalog.load_table(identifier)
+ raise AssertionError()
+ except NoSuchTableError:
+ assert True
+ finally:
+ test_catalog.close()
 
 
 def test_drop_namespace(mocker: MockFixture, gcp_dataset_name: str) -> None:
@@ -116,11 +122,14 @@ def test_drop_namespace(mocker: MockFixture, gcp_dataset_name: str) -> None:
  catalog_name = "test_catalog"
  test_catalog = BigQueryMetastoreCatalog(catalog_name, **{"gcp.bigquery.project-id": "alexstephen-test-1"})
 
- test_catalog.drop_namespace(gcp_dataset_name)
- client_mock.delete_dataset.assert_called_once()
- args, _ = client_mock.delete_dataset.call_args
- assert isinstance(args[0], Dataset)
- assert args[0].dataset_id == gcp_dataset_name
+ try:
+ test_catalog.drop_namespace(gcp_dataset_name)
+ client_mock.delete_dataset.assert_called_once()
+ args, _ = client_mock.delete_dataset.call_args
+ assert isinstance(args[0], Dataset)
+ assert args[0].dataset_id == gcp_dataset_name
+ finally:
+ test_catalog.close()
 
 
 def test_list_tables(mocker: MockFixture, gcp_dataset_name: str) -> None:
@@ -151,14 +160,19 @@ def test_list_tables(mocker: MockFixture, gcp_dataset_name: str) -> None:
  catalog_name = "test_catalog"
  test_catalog = BigQueryMetastoreCatalog(catalog_name, **{"gcp.bigquery.project-id": "my-project"})
 
- tables = test_catalog.list_tables(gcp_dataset_name)
+ try:
+ tables = test_catalog.list_tables(gcp_dataset_name)
 
- # Assert that all tables returned by client.list_tables are listed
- assert len(tables) == 2
- assert (gcp_dataset_name, "iceberg_table_A") in tables
- assert (gcp_dataset_name, "iceberg_table_B") in tables
+# Assert that all tables returned by client.list_tables are listed
+assert len(tables) == 2
+assert (gcp_dataset_name, "iceberg_table_A") in tables
+assert (gcp_dataset_name, "iceberg_table_B") in tables
 
- client_mock.list_tables.assert_called_once_with(dataset=DatasetReference(project="my-project", dataset_id=gcp_dataset_name))
+ client_mock.list_tables.assert_called_once_with(
+ dataset=DatasetReference(project="my-project", dataset_id=gcp_dataset_name)
+ )
+ finally:
+ test_catalog.close()
 
 
 def test_list_namespaces(mocker: MockFixture) -> None:
@@ -173,8 +187,11 @@ def test_list_namespaces(mocker: MockFixture) -> None:
  catalog_name = "test_catalog"
  test_catalog = BigQueryMetastoreCatalog(catalog_name, **{"gcp.bigquery.project-id": "my-project"})
 
- namespaces = test_catalog.list_namespaces()
- assert len(namespaces) == 2
- assert ("dataset1",) in namespaces
- assert ("dataset2",) in namespaces
- client_mock.list_datasets.assert_called_once()
+ try:
+ namespaces = test_catalog.list_namespaces()
+ assert len(namespaces) == 2
+ assert ("dataset1",) in namespaces
+ assert ("dataset2",) in namespaces
+ client_mock.list_datasets.assert_called_once()
+ finally:
+ test_catalog.close()

tests/catalog/test_sql.py

@@ -48,6 +48,11 @@ def catalog_name() -> str:
 
 @pytest.fixture(scope="module")
 def catalog_memory(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
+ """In-memory SQLite catalog fixture.
+
+ Calls catalog.close() during teardown to properly dispose SQLAlchemy
+ connection pools and prevent Python 3.13 ResourceWarning. See #2530.
+ """
  props = {
  "uri": "sqlite:///:memory:",
  "warehouse": f"file://{warehouse}",
@@ -56,10 +61,16 @@ def catalog_memory(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog,
  catalog.create_tables()
  yield catalog
  catalog.destroy_tables()
+ catalog.close()
 
 
 @pytest.fixture(scope="module")
 def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
+ """File-based SQLite catalog fixture.
+
+ Calls catalog.close() during teardown to properly dispose SQLAlchemy
+ connection pools and prevent Python 3.13 ResourceWarning. See #2530.
+ """
  props = {
  "uri": f"sqlite:////{warehouse}/sql-catalog",
  "warehouse": f"file://{warehouse}",
@@ -68,6 +79,7 @@ def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog,
  catalog.create_tables()
  yield catalog
  catalog.destroy_tables()
+ catalog.close()
 
 
 @pytest.fixture(scope="module")
@@ -76,8 +88,10 @@ def catalog_uri(warehouse: Path) -> str:
 
 
 @pytest.fixture(scope="module")
-def alchemy_engine(catalog_uri: str) -> Engine:
- return create_engine(catalog_uri)
+def alchemy_engine(catalog_uri: str) -> Generator[Engine, None, None]:
+ engine = create_engine(catalog_uri)
+ yield engine
+ engine.dispose()
 
 
 def test_creation_with_no_uri(catalog_name: str) -> None:
@@ -103,10 +117,13 @@ def test_creation_with_echo_parameter(catalog_name: str, warehouse: Path) -> Non
  if echo_param is not None:
  props["echo"] = echo_param
  catalog = SqlCatalog(catalog_name, **props)
- assert catalog.engine._echo == expected_echo_value, (
- f"Assertion failed: expected echo value {expected_echo_value}, "
- f"but got {catalog.engine._echo}. For echo_param={echo_param}"
- )
+ try:
+ assert catalog.engine._echo == expected_echo_value, (
+ f"Assertion failed: expected echo value {expected_echo_value}, "
+ f"but got {catalog.engine._echo}. For echo_param={echo_param}"
+ )
+ finally:
+ catalog.close()
 
 
 def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Path) -> None:
@@ -127,24 +144,29 @@ def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Pat
  props["pool_pre_ping"] = pool_pre_ping_param
 
  catalog = SqlCatalog(catalog_name, **props)
- assert catalog.engine.pool._pre_ping == expected_pool_pre_ping_value, (
- f"Assertion failed: expected pool_pre_ping value {expected_pool_pre_ping_value}, "
- f"but got {catalog.engine.pool._pre_ping}. For pool_pre_ping_param={pool_pre_ping_param}"
- )
+ try:
+ assert catalog.engine.pool._pre_ping == expected_pool_pre_ping_value, (
+ f"Assertion failed: expected pool_pre_ping value {expected_pool_pre_ping_value}, "
+ f"but got {catalog.engine.pool._pre_ping}. For pool_pre_ping_param={pool_pre_ping_param}"
+ )
+ finally:
+ catalog.close()
 
 
 def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None:
- assert isinstance(
- load_catalog(
- catalog_name,
- **{
- "py-catalog-impl": "pyiceberg.catalog.sql.SqlCatalog",
- "uri": f"sqlite:////{warehouse}/sql-catalog",
- "warehouse": f"file://{warehouse}",
- },
- ),
- SqlCatalog,
+ catalog = load_catalog(
+ catalog_name,
+ **{
+ "py-catalog-impl": "pyiceberg.catalog.sql.SqlCatalog",
+ "uri": f"sqlite:////{warehouse}/sql-catalog",
+ "warehouse": f"file://{warehouse}",
+ },
  )
+ try:
+ assert isinstance(catalog, SqlCatalog)
+ finally:
+ if isinstance(catalog, SqlCatalog):
+ catalog.close()
 
 
 def confirm_no_tables_exist(alchemy_engine: Engine) -> None:
@@ -182,7 +204,10 @@ def load_catalog_for_catalog_table_creation(catalog_name: str, catalog_uri: str)
 def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
  confirm_no_tables_exist(alchemy_engine)
  catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
- confirm_all_tables_exist(catalog)
+ try:
+ confirm_all_tables_exist(catalog)
+ finally:
+ catalog.close()
 
 
 def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
@@ -194,7 +219,10 @@ def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: s
  assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]
 
  catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
- confirm_all_tables_exist(catalog)
+ try:
+ confirm_all_tables_exist(catalog)
+ finally:
+ catalog.close()
 
 
 def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
@@ -207,7 +235,10 @@ def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: s
  assert c in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]
 
  catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
- confirm_all_tables_exist(catalog)
+ try:
+ confirm_all_tables_exist(catalog)
+ finally:
+ catalog.close()
 
 
 class TestSqlCatalogClose:
@@ -261,3 +292,22 @@ def test_sql_catalog_multiple_close_calls(self, catalog_sqlite: SqlCatalog) -> N
 
  # Second close should not raise an exception
  catalog_sqlite.close()
+
+ def test_catalog_fixture_closes_connections(self, warehouse: Path) -> None:
+ """Regression test: verify SqlCatalog properly closes all SQLAlchemy
+ connections to prevent Python 3.13 ResourceWarning about unclosed
+ database connections. See: apache/iceberg-python#2530"""
+ import gc
+ import warnings
+
+ props = {
+ "uri": f"sqlite:////{warehouse}/test-regression-2530",
+ "warehouse": f"file://{warehouse}",
+ }
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", ResourceWarning)
+ catalog = SqlCatalog("regression_test", **props)
+ catalog.create_tables()
+ catalog.destroy_tables()
+ catalog.close()
+ gc.collect() # Force GC to catch any unclosed connections