@@ -238,7 +238,10 @@ bigquery-test: guard-BIGQUERY_KEYFILE engine-bigquery-install
 	pytest -n auto -m "bigquery" --reruns 3 --junitxml=test-results/junit-bigquery.xml
 
 databricks-test: guard-DATABRICKS_CATALOG guard-DATABRICKS_SERVER_HOSTNAME guard-DATABRICKS_HTTP_PATH guard-DATABRICKS_CONNECT_VERSION engine-databricks-install
-	$(PIP) install 'databricks-connect==${DATABRICKS_CONNECT_VERSION}'
+	# Connect 15.3.x requires numpy<2 and pyarrow>=4 (no upper bound). Cap PyArrow
+	# in the same resolve that downgrades NumPy, or PyArrow 25+ is left installed
+	# and cannot be imported.
+	$(PIP) install 'databricks-connect==${DATABRICKS_CONNECT_VERSION}' 'pyarrow<25'
 	pytest -n auto -m "databricks" --reruns 3 --junitxml=test-results/junit-databricks.xml
 
 redshift-test: guard-REDSHIFT_HOST guard-REDSHIFT_USER guard-REDSHIFT_PASSWORD guard-REDSHIFT_DATABASE engine-redshift-install
 
@@ -757,14 +757,32 @@ def close(self) -> t.Any:
     def get_table_last_modified_ts(self, table_names: t.List[TableName]) -> t.List[int]:
         from sqlmesh.utils.date import to_timestamp
 
-        num_tables = len(table_names)
-
-        query = "SELECT LAST_ALTERED FROM INFORMATION_SCHEMA.TABLES WHERE"
-        for i, table_name in enumerate(table_names):
+        # INFORMATION_SCHEMA only contains the objects of the database it belongs to, so the tables
+        # are grouped by database and each group is looked up in its own database's INFORMATION_SCHEMA.
+        # The groups are keyed on the catalog identifier rather than its name, because quoting matters
+        # in Snowflake: unquoted raw is the database RAW, while quoted "raw" is a different database
+        catalogs_to_tables: t.Dict[t.Optional[exp.Identifier], t.List[exp.Table]] = {}
+        for table_name in table_names:
             table = exp.to_table(table_name)
-            query += f"""(TABLE_NAME = '{table.name}' AND TABLE_SCHEMA = '{table.db}' AND TABLE_CATALOG = '{table.catalog}')"""
-            if i < num_tables - 1:
-                query += " OR "
+            catalogs_to_tables.setdefault(table.args.get("catalog"), []).append(table)
+
+        results = []
+        for catalog, tables in catalogs_to_tables.items():
+            query = (
+                exp.select("LAST_ALTERED")
+                .from_(exp.table_("TABLES", db="INFORMATION_SCHEMA", catalog=catalog))
+                .where(
+                    exp.or_(
+                        *(
+                            exp.and_(
+                                exp.column("TABLE_NAME").eq(exp.Literal.string(table.name)),
+                                exp.column("TABLE_SCHEMA").eq(exp.Literal.string(table.db)),
+                            )
+                            for table in tables
+                        )
+                    )
+                )
+            )
+            results.extend(self.fetchall(query))
 
-        result = self.fetchall(query)
-        return [to_timestamp(row[0]) for row in result]
+        return [to_timestamp(row[0]) for row in results]
@@ -65,26 +65,40 @@ def run(self) -> t.Tuple[t.List[NodeExecutionFailedError[H]], t.List[H]]:
         return self._node_errors, self._skipped_nodes
 
     def _process_node(self, node: H, executor: Executor) -> None:
+        with self._unprocessed_nodes_lock:
+            if self._finished_future.done():
+                return
+
         try:
             self.fn(node)
 
             with self._unprocessed_nodes_lock:
+                if self._finished_future.done():
+                    return
                 self._unprocessed_nodes_num -= 1
                 self._submit_next_nodes(executor, node)
         except Exception as ex:
             error = NodeExecutionFailedError(node)
             error.__cause__ = ex
 
-            if self.raise_on_error:
-                self._finished_future.set_exception(error)
-                return
-
             with self._unprocessed_nodes_lock:
+                if self._finished_future.done():
+                    return
+                if self.raise_on_error:
+                    self._finished_future.set_exception(error)
+                    return
                 self._unprocessed_nodes_num -= 1
                 self._node_errors.append(error)
                 self._skip_next_nodes(node)
+        except BaseException as ex:
+            with self._unprocessed_nodes_lock:
+                if not self._finished_future.done():
+                    self._finished_future.set_exception(ex)
 
     def _submit_next_nodes(self, executor: Executor, processed_node: t.Optional[H] = None) -> None:
+        if self._finished_future.done():
+            return
+
         if not self._unprocessed_nodes_num:
             self._finished_future.set_result(None)
             return
@@ -101,6 +115,9 @@ def _submit_next_nodes(self, executor: Executor, processed_node: t.Optional[H] =
             executor.submit(self._process_node, submitted_node, executor)
 
     def _skip_next_nodes(self, parent: H) -> None:
+        if self._finished_future.done():
+            return
+
         if not self._unprocessed_nodes_num:
             self._finished_future.set_result(None)
             return
@@ -267,9 +284,11 @@ def concurrent_apply_to_values(
 
     def _process_value(value: A, index: int) -> None:
         try:
-            futures[index].set_result(fn(value))
-        except Exception as ex:
+            result = fn(value)
+        except BaseException as ex:
             futures[index].set_exception(ex)
+        else:
+            futures[index].set_result(result)
 
     with ThreadPoolExecutor(max_workers=tasks_num) as pool:
         for index, value in enumerate(values):
 
@@ -1,4 +1,5 @@
 import typing as t
+from datetime import datetime, timezone
 
 import pandas as pd  # noqa: TID253
 import pytest
@@ -14,6 +15,7 @@
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.model.definition import SqlModel
 from sqlmesh.core.node import IntervalUnit
+from sqlmesh.utils.date import to_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils import optional_import
 from tests.core.engine_adapter import to_sql_calls
@@ -73,6 +75,38 @@ def test_get_data_objects_lowercases_columns(
     assert data_object.clustering_key == "ID"
 
 
+def test_get_table_last_modified_ts_multiple_catalogs(
+    snowflake_mocked_engine_adapter: SnowflakeEngineAdapter,
+) -> None:
+    adapter = snowflake_mocked_engine_adapter
+    adapter.cursor.fetchall.side_effect = [
+        [
+            (datetime(2025, 1, 1, tzinfo=timezone.utc),),
+            (datetime(2025, 1, 2, tzinfo=timezone.utc),),
+        ],
+        [(datetime(2025, 1, 3, tzinfo=timezone.utc),)],
+    ]
+
+    result = adapter.get_table_last_modified_ts(
+        [
+            '"ANALYTICS"."STAGING"."ORDERS"',
+            '"RAW"."SHOPIFY"."ORDERS"',
+            '"ANALYTICS"."STAGING"."CUSTOMERS"',
+        ]
+    )
+
+    # INFORMATION_SCHEMA only covers the database it belongs to, so each database is queried separately
+    assert to_sql_calls(adapter) == [
+        """SELECT LAST_ALTERED FROM "ANALYTICS".INFORMATION_SCHEMA.TABLES WHERE (TABLE_NAME = 'ORDERS' AND TABLE_SCHEMA = 'STAGING') OR (TABLE_NAME = 'CUSTOMERS' AND TABLE_SCHEMA = 'STAGING')""",
+        """SELECT LAST_ALTERED FROM "RAW".INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'ORDERS' AND TABLE_SCHEMA = 'SHOPIFY'""",
+    ]
+    assert result == [
+        to_timestamp("2025-01-01"),
+        to_timestamp("2025-01-02"),
+        to_timestamp("2025-01-03"),
+    ]
+
+
 @pytest.mark.parametrize(
     "current_warehouse, current_warehouse_exp, configured_warehouse, configured_warehouse_exp, should_change",
     [
@@ -183,6 +217,29 @@ def test_alter_session_parameters_are_not_quoted(make_mocked_engine_adapter: t.C
     ]
 
 
+def test_get_table_last_modified_ts_quoted_and_unquoted_catalogs(
+    snowflake_mocked_engine_adapter: SnowflakeEngineAdapter,
+) -> None:
+    # unquoted raw resolves to the database RAW, quoted "raw" is a different database,
+    # so the two must be looked up in two different information schemas
+    adapter = snowflake_mocked_engine_adapter
+    adapter.cursor.fetchall.side_effect = [
+        [(datetime(2025, 1, 1, tzinfo=timezone.utc),)],
+        [(datetime(2025, 1, 2, tzinfo=timezone.utc),)],
+    ]
+
+    result = adapter.get_table_last_modified_ts(['"raw".shopify.orders', "raw.shopify.customers"])
+
+    assert result == [
+        to_timestamp(datetime(2025, 1, 1, tzinfo=timezone.utc)),
+        to_timestamp(datetime(2025, 1, 2, tzinfo=timezone.utc)),
+    ]
+    assert to_sql_calls(adapter) == [
+        """SELECT LAST_ALTERED FROM "raw".INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'orders' AND TABLE_SCHEMA = 'shopify'""",
+        """SELECT LAST_ALTERED FROM raw.INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'customers' AND TABLE_SCHEMA = 'shopify'""",
+    ]
+
+
 def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
 
 
@@ -2609,19 +2609,14 @@ def test_requirements(copy_to_temp_path: t.Callable):
         "dev", no_prompts=True, skip_tests=True, skip_backfill=True, auto_apply=True
     ).environment
     requirements = {"ipywidgets", "numpy", "pandas", "test_package"}
-    if IS_WINDOWS:
-        requirements.add("pendulum")
     assert environment.requirements["pandas"] == "2.2.2"
     assert set(environment.requirements) == requirements
 
     context._requirements = {"numpy": "2.1.2", "pandas": "2.2.1"}
     context._excluded_requirements = {"ipywidgets", "ruamel.yaml", "ruamel.yaml.clib"}
     diff = context.plan_builder("dev", skip_tests=True, skip_backfill=True).build().context_diff
     assert set(diff.previous_requirements) == requirements
-    reqs = {"numpy", "pandas"}
-    if IS_WINDOWS:
-        reqs.add("pendulum")
-    assert set(diff.requirements) == reqs
+    assert set(diff.requirements) == {"numpy", "pandas"}
 
 
 def test_deactivate_automatic_requirement_inference(copy_to_temp_path: t.Callable):