Introduce TableLastModifiedMetadataBatch and implement BaseAdapter.calculate_freshness_from_metadata_batch

dbt/adapters/base/impl.py

@@ -423,7 +423,9 @@ def _get_cache_schemas(self, relation_configs: Iterable[RelationConfig]) -> Set[
         populate.
         """
         return {
-            self.Relation.create_from(quoting=self.config, relation_config=relation_config).without_identifier()
+            self.Relation.create_from(
+                quoting=self.config, relation_config=relation_config
+            ).without_identifier()
             for relation_config in relation_configs
         }
 
@@ -1285,46 +1287,92 @@ def calculate_freshness(
         }
         return adapter_response, freshness
 
+    def calculate_freshness_from_metadata_batch(
+        self,
+        sources: List[BaseRelation],
+        macro_resolver: Optional[MacroResolverProtocol] = None,
+    ) -> Tuple[Optional[AdapterResponse], Dict[BaseRelation, FreshnessResponse]]:
+        # Track schema, identifiers of sources for lookup from batch query
+        schema_identifier_to_source = {
+            (
+                source.path.get_lowered_part(ComponentName.Schema),
+                source.path.get_lowered_part(ComponentName.Identifier),
+            ): source
+            for source in sources
+        }
+
+        # Group metadata sources by information schema -- one query per information schema will be necessary
+        information_schema_to_metadata_sources: Dict[InformationSchema, List[BaseRelation]] = {}
+        for source in sources:
+            information_schema = source.information_schema_only()
+            if information_schema not in information_schema_to_metadata_sources:
+                information_schema_to_metadata_sources[information_schema] = [source]
+            else:
+                information_schema_to_metadata_sources[information_schema].append(source)
+
+        freshness_responses: Dict[BaseRelation, FreshnessResponse] = {}
+        for (
+            information_schema,
+            sources_for_information_schema,
+        ) in information_schema_to_metadata_sources.items():
+            result = self.execute_macro(
+                GET_RELATION_LAST_MODIFIED_MACRO_NAME,
+                kwargs={
+                    "information_schema": information_schema,
+                    "relations": sources_for_information_schema,
+                },
+                macro_resolver=macro_resolver,
+            )
+            adapter_response, table = result.response, result.table  # type: ignore[attr-defined]
+
+            for row in table:
+                try:
+                    last_modified_val = get_column_value_uncased("last_modified", row)
+                    snapshotted_at_val = get_column_value_uncased("snapshotted_at", row)
+                    identifier = get_column_value_uncased("identifier", row)
+                    schema = get_column_value_uncased("schema", row)
+                except Exception:
+                    raise MacroResultError(GET_RELATION_LAST_MODIFIED_MACRO_NAME, table)
+
+                freshness_response = self._create_freshness_response(
+                    last_modified_val, snapshotted_at_val
+                )
+                source_relation_for_result = schema_identifier_to_source[
+                    (schema.lower(), identifier.lower())
+                ]
+                freshness_responses[source_relation_for_result] = freshness_response
+
+        return adapter_response, freshness_responses
+
     def calculate_freshness_from_metadata(
         self,
         source: BaseRelation,
         macro_resolver: Optional[MacroResolverProtocol] = None,
     ) -> Tuple[Optional[AdapterResponse], FreshnessResponse]:
-        kwargs: Dict[str, Any] = {
-            "information_schema": source.information_schema_only(),
-            "relations": [source],
-        }
-        result = self.execute_macro(
-            GET_RELATION_LAST_MODIFIED_MACRO_NAME,
-            kwargs=kwargs,
+        adapter_response, freshness_responses = self.calculate_freshness_from_metadata_batch(
+            sources=[source],
             macro_resolver=macro_resolver,
         )
-        adapter_response, table = result.response, result.table  # type: ignore[attr-defined]
-
-        try:
-            row = table[0]
-            last_modified_val = get_column_value_uncased("last_modified", row)
-            snapshotted_at_val = get_column_value_uncased("snapshotted_at", row)
-        except Exception:
-            raise MacroResultError(GET_RELATION_LAST_MODIFIED_MACRO_NAME, table)
+        return adapter_response, list(freshness_responses.values())[0]
 
-        if last_modified_val is None:
+    def _create_freshness_response(
+        self, last_modified: Optional[datetime], snapshotted_at: Optional[datetime]
+    ) -> FreshnessResponse:
+        if last_modified is None:
             # Interpret missing value as "infinitely long ago"
             max_loaded_at = datetime(1, 1, 1, 0, 0, 0, tzinfo=pytz.UTC)
         else:
-            max_loaded_at = _utc(last_modified_val, None, "last_modified")
-
-        snapshotted_at = _utc(snapshotted_at_val, None, "snapshotted_at")
+            max_loaded_at = _utc(last_modified, None, "last_modified")
 
+        snapshotted_at = _utc(snapshotted_at, None, "snapshotted_at")
         age = (snapshotted_at - max_loaded_at).total_seconds()
-
         freshness: FreshnessResponse = {
             "max_loaded_at": max_loaded_at,
             "snapshotted_at": snapshotted_at,
             "age": age,
         }
 
-        return adapter_response, freshness
+        return freshness
 
     def pre_model_hook(self, config: Mapping[str, Any]) -> Any:
         """A hook for running some operation before the model materialization

dbt/adapters/capability.py

@@ -13,6 +13,9 @@ class Capability(str, Enum):
     TableLastModifiedMetadata = "TableLastModifiedMetadata"
     """Indicates support for determining the time of the last table modification by querying database metadata."""
 
+    TableLastModifiedMetadataBatch = "TableLastModifiedMetadataBatch"
+    """Indicates support for determining the time of the last table modification by querying database metadata in batch."""
+
 
 class Support(str, Enum):
     Unknown = "Unknown"