diff --git a/dbt/adapters/spark/impl.py b/dbt/adapters/spark/impl.py
index b26040608..22e936a52 100644
--- a/dbt/adapters/spark/impl.py
+++ b/dbt/adapters/spark/impl.py
@@ -385,32 +385,31 @@ def get_catalog_by_relations(
         self, manifest: Manifest, relations: Set[BaseRelation]
     ) -> Tuple[agate.Table, List[Exception]]:
         info_schemas = {r.information_schema() for r in relations}
-        if len(info_schemas) > 1:
-            raise dbt.exceptions.CompilationError(
-                f"Expected only one database in get_catalog_by_relations, found {list(info_schemas)}"
-            )
+
+        if len(info_schemas) == 0:
+            return catch_as_completed([])
+
         elif len(info_schemas) == 1:
             info_schema = info_schemas.pop()
+            with executor(self.config) as tpe:
+                return catch_as_completed(
+                    [
+                        tpe.submit_connected(
+                            self,
+                            info_schema.identifier,
+                            self._get_one_catalog_by_relations,
+                            info_schema,
+                            relations,
+                            manifest,
+                        )
+                    ]
+                )
+
         else:
             raise dbt.exceptions.CompilationError(
-                "Expected a database in get_catalog_by_relations, found none"
+                f"Expected only one database in get_catalog_by_relations, found {list(info_schemas)}"
             )
 
-        with executor(self.config) as tpe:
-            # futures: List[Future[agate.Table]] = []
-            futures = [
-                tpe.submit_connected(
-                    self,
-                    "information_schema",
-                    self._get_one_catalog_by_relations,
-                    info_schema,
-                    relations,
-                    manifest,
-                )
-            ]
-            catalogs, exceptions = catch_as_completed(futures)
-        return catalogs, exceptions
-
     def _get_one_catalog(
         self,
         information_schema: InformationSchema,