diff --git a/metricflow/dataflow/builder/dataflow_plan_builder.py b/metricflow/dataflow/builder/dataflow_plan_builder.py
index 2b2d30fd06..134ce7584e 100644
--- a/metricflow/dataflow/builder/dataflow_plan_builder.py
+++ b/metricflow/dataflow/builder/dataflow_plan_builder.py
@@ -60,6 +60,7 @@
 from metricflow.errors.errors import UnableToSatisfyQueryError
 from metricflow.filters.time_constraint import TimeRangeConstraint
 from metricflow.mf_logging.pretty_print import mf_pformat
+from metricflow.mf_logging.runtime import log_runtime
 from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
 from metricflow.plan_conversion.column_resolver import DunderColumnAssociationResolver
 from metricflow.plan_conversion.node_processor import PreJoinNodeProcessor
@@ -149,6 +150,7 @@ def __init__(  # noqa: D
             else node_output_resolver
         )
 
+    @log_runtime()
     def build_plan(
         self,
         query_spec: MetricFlowQuerySpec,
@@ -583,6 +585,7 @@ def _build_metrics_output_node(
 
         return CombineAggregatedOutputsNode(parent_nodes=output_nodes)
 
+    @log_runtime()
     def build_plan_for_distinct_values(self, query_spec: MetricFlowQuerySpec) -> DataflowPlan:
         """Generate a plan that would get the distinct values of a linkable instance.
 
diff --git a/metricflow/mf_logging/runtime.py b/metricflow/mf_logging/runtime.py
new file mode 100644
index 0000000000..57f036ddc2
--- /dev/null
+++ b/metricflow/mf_logging/runtime.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import logging
+import time
+from typing import Callable, TypeVar
+
+from typing_extensions import ParamSpec
+
+logger = logging.getLogger(__name__)
+
+ReturnType = TypeVar("ReturnType")
+ParametersType = ParamSpec("ParametersType")
+
+
+def log_runtime(
+    runtime_warning_threshold: float = 3.0,
+) -> Callable[[Callable[ParametersType, ReturnType]], Callable[ParametersType, ReturnType]]:
+    """Logs how long a function took to run.
+
+    If the runtime exceeds runtime_warning_threshold, then a warning is logged.
+    """
+
+    def decorator(wrapped_function: Callable[ParametersType, ReturnType]) -> Callable[ParametersType, ReturnType]:
+        def _inner(*args: ParametersType.args, **kwargs: ParametersType.kwargs) -> ReturnType:
+            # __qualname__ includes the path like MyClass.my_function
+            function_name = f"{wrapped_function.__qualname__}()"
+            start_time = time.time()
+            logger.info(f"Starting {function_name}")
+
+            try:
+                result = wrapped_function(*args, **kwargs)
+            finally:
+                runtime = time.time() - start_time
+                logger.info(f"Finished {function_name} in {runtime:.1f}s")
+                if runtime > runtime_warning_threshold:
+                    logger.warning(f"{function_name} is slow with a runtime of {runtime:.1f}s")
+
+            return result
+
+        return _inner
+
+    return decorator
diff --git a/metricflow/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py b/metricflow/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
index 00039137b9..a47fc445d0 100644
--- a/metricflow/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
+++ b/metricflow/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
@@ -11,6 +11,7 @@
 from dbt_semantic_interfaces.protocols import WhereFilter, WhereFilterIntersection
 from typing_extensions import override
 
+from metricflow.mf_logging.runtime import log_runtime
 from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
 from metricflow.naming.object_builder_str import ObjectBuilderNameConverter
 from metricflow.query.group_by_item.candidate_push_down.push_down_visitor import DagTraversalPathTracker
@@ -70,6 +71,7 @@ def __init__(  # noqa: D
         self._manifest_lookup = manifest_lookup
         self._resolution_dag = resolution_dag
 
+    @log_runtime()
     def resolve_lookup(self) -> FilterSpecResolutionLookUp:
         """Find all where filters and return a lookup that provides the specs for the included group-by-items."""
         visitor = _ResolveWhereFilterSpecVisitor(manifest_lookup=self._manifest_lookup)
diff --git a/metricflow/query/query_parser.py b/metricflow/query/query_parser.py
index 51a7e9e0cb..6990b715e1 100644
--- a/metricflow/query/query_parser.py
+++ b/metricflow/query/query_parser.py
@@ -19,6 +19,7 @@
 from metricflow.filters.time_constraint import TimeRangeConstraint
 from metricflow.mf_logging.formatting import indent
 from metricflow.mf_logging.pretty_print import mf_pformat
+from metricflow.mf_logging.runtime import log_runtime
 from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
 from metricflow.naming.dunder_scheme import DunderNamingScheme
 from metricflow.naming.metric_scheme import MetricNamingScheme
@@ -295,6 +296,7 @@ def _raise_exception_if_there_are_errors(
 
         raise InvalidQueryException(self.generate_error_message(input_to_issue_set=input_to_issue_set))
 
+    @log_runtime()
     def parse_and_validate_query(
         self,
         metric_names: Optional[Sequence[str]] = None,
diff --git a/metricflow/query/query_resolver.py b/metricflow/query/query_resolver.py
index a36566dd5c..46fe1324d4 100644
--- a/metricflow/query/query_resolver.py
+++ b/metricflow/query/query_resolver.py
@@ -8,6 +8,7 @@
 
 from metricflow.dag.dag_to_text import dag_as_text
 from metricflow.mf_logging.pretty_print import mf_pformat
+from metricflow.mf_logging.runtime import log_runtime
 from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
 from metricflow.naming.metric_scheme import MetricNamingScheme
 from metricflow.query.group_by_item.filter_spec_resolution.filter_spec_lookup import FilterSpecResolutionLookUp
@@ -319,6 +320,7 @@ def _build_filter_spec_lookup(
 
         return where_filter_spec_resolver.resolve_lookup()
 
+    @log_runtime()
     def resolve_query(self, resolver_input_for_query: ResolverInputForQuery) -> MetricFlowQueryResolution:
         """Resolve the query into specs that can be passed into the next stage in query processing."""
         metric_inputs = resolver_input_for_query.metric_inputs