diff --git a/extra-hatch-configuration/requirements.txt b/extra-hatch-configuration/requirements.txt
index 078b5a2b4e..e2b855c60b 100644
--- a/extra-hatch-configuration/requirements.txt
+++ b/extra-hatch-configuration/requirements.txt
@@ -1,5 +1,5 @@
 Jinja2>=3.1.3
-dbt-semantic-interfaces==0.7.2
+dbt-semantic-interfaces==0.8.1
 more-itertools>=8.10.0, <10.2.0
 pydantic>=1.10.0, <3.0
 tabulate>=0.8.9
diff --git a/metricflow-semantics/extra-hatch-configuration/requirements.txt b/metricflow-semantics/extra-hatch-configuration/requirements.txt
index 3f64b7014e..85a8abdd38 100644
--- a/metricflow-semantics/extra-hatch-configuration/requirements.txt
+++ b/metricflow-semantics/extra-hatch-configuration/requirements.txt
@@ -1,7 +1,7 @@
 # Always support a range of production DSI versions capped at the next breaking version in metricflow-semantics.
 # This allows us to sync new, non-breaking changes to dbt-core without getting a version mismatch in dbt-mantle,
 # which depends on a specific commit of DSI.
-dbt-semantic-interfaces>=0.7.2, <0.8.0
+dbt-semantic-interfaces>=0.8.1, <0.9.0
 graphviz>=0.18.2, <0.21
 python-dateutil>=2.9.0, <2.10.0
 rapidfuzz>=3.0, <4.0
diff --git a/metricflow-semantics/metricflow_semantics/errors/custom_grain_not_supported.py b/metricflow-semantics/metricflow_semantics/errors/custom_grain_not_supported.py
new file mode 100644
index 0000000000..5ac1cb56c3
--- /dev/null
+++ b/metricflow-semantics/metricflow_semantics/errors/custom_grain_not_supported.py
@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
+
+
+def error_if_not_standard_grain(input_granularity: str, context: Optional[str] = None) -> TimeGranularity:
+    """Cast input grainularity string to TimeGranularity, otherwise error.
+
+    TODO: Not needed once, custom grain is supported for most things.
+    """
+    try:
+        time_grain = TimeGranularity(input_granularity)
+    except ValueError:
+        error_msg = f"Received a non-standard time granularity, which is not supported at the moment, received: {input_granularity}."
+        if context:
+            error_msg += f"\nContext: {context}"
+        raise ValueError(error_msg)
+    return time_grain
diff --git a/metricflow-semantics/metricflow_semantics/naming/dunder_scheme.py b/metricflow-semantics/metricflow_semantics/naming/dunder_scheme.py
index df3e66879e..ebef953fdd 100644
--- a/metricflow-semantics/metricflow_semantics/naming/dunder_scheme.py
+++ b/metricflow-semantics/metricflow_semantics/naming/dunder_scheme.py
@@ -23,7 +23,7 @@
 class DunderNamingScheme(QueryItemNamingScheme):
     """A naming scheme using the dundered name syntax.
 
-    TODO: Consolidate with StructuredLinkableSpecName / DunderedNameFormatter.
+    TODO: Consolidate with StructuredLinkableSpecName.
     """
 
     _INPUT_REGEX = re.compile(r"\A[a-z]([a-z0-9_])*[a-z0-9]\Z")
@@ -52,7 +52,7 @@ def input_str(self, instance_spec: InstanceSpec) -> Optional[str]:
 
     @override
     def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> EntityLinkPattern:
-        if not self.input_str_follows_scheme(input_str):
+        if not self.input_str_follows_scheme(input_str, semantic_manifest_lookup=semantic_manifest_lookup):
             raise ValueError(f"{repr(input_str)} does not follow this scheme.")
 
         input_str = input_str.lower()
@@ -119,7 +119,7 @@ def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifes
         )
 
     @override
-    def input_str_follows_scheme(self, input_str: str) -> bool:
+    def input_str_follows_scheme(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> bool:
         # This naming scheme is case-insensitive.
         input_str = input_str.lower()
         if DunderNamingScheme._INPUT_REGEX.match(input_str) is None:
diff --git a/metricflow-semantics/metricflow_semantics/naming/linkable_spec_name.py b/metricflow-semantics/metricflow_semantics/naming/linkable_spec_name.py
index ecca286ccb..1028c322f1 100644
--- a/metricflow-semantics/metricflow_semantics/naming/linkable_spec_name.py
+++ b/metricflow-semantics/metricflow_semantics/naming/linkable_spec_name.py
@@ -4,6 +4,7 @@
 from functools import lru_cache
 from typing import Optional, Sequence, Tuple
 
+from dbt_semantic_interfaces.references import EntityReference
 from dbt_semantic_interfaces.type_enums.date_part import DatePart
 from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
 
@@ -106,6 +107,11 @@ def date_part_suffix(date_part: DatePart) -> str:
         """Suffix used for names with a date_part."""
         return f"extract_{date_part.value}"
 
+    @property
+    def entity_links(self) -> Tuple[EntityReference, ...]:
+        """Returns the entity link references."""
+        return tuple(EntityReference(entity_link_name.lower()) for entity_link_name in self.entity_link_names)
+
     @property
     def granularity_free_qualified_name(self) -> str:
         """Renders the qualified name without the granularity suffix.
diff --git a/metricflow-semantics/metricflow_semantics/naming/metric_scheme.py b/metricflow-semantics/metricflow_semantics/naming/metric_scheme.py
index fcd9fe5ac9..a19407f5d6 100644
--- a/metricflow-semantics/metricflow_semantics/naming/metric_scheme.py
+++ b/metricflow-semantics/metricflow_semantics/naming/metric_scheme.py
@@ -28,12 +28,12 @@ def input_str(self, instance_spec: InstanceSpec) -> Optional[str]:
     @override
     def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> MetricSpecPattern:
         input_str = input_str.lower()
-        if not self.input_str_follows_scheme(input_str):
+        if not self.input_str_follows_scheme(input_str, semantic_manifest_lookup=semantic_manifest_lookup):
             raise RuntimeError(f"{repr(input_str)} does not follow this scheme.")
         return MetricSpecPattern(metric_reference=MetricReference(element_name=input_str))
 
     @override
-    def input_str_follows_scheme(self, input_str: str) -> bool:
+    def input_str_follows_scheme(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> bool:
         # TODO: Use regex.
         return True
 
diff --git a/metricflow-semantics/metricflow_semantics/naming/naming_scheme.py b/metricflow-semantics/metricflow_semantics/naming/naming_scheme.py
index f9110c01b8..f78cf72ae9 100644
--- a/metricflow-semantics/metricflow_semantics/naming/naming_scheme.py
+++ b/metricflow-semantics/metricflow_semantics/naming/naming_scheme.py
@@ -42,7 +42,7 @@ def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifes
         pass
 
     @abstractmethod
-    def input_str_follows_scheme(self, input_str: str) -> bool:
+    def input_str_follows_scheme(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> bool:
         """Returns true if the given input string follows this naming scheme.
 
         Consider adding a structured result that indicates why it does not match the scheme.
diff --git a/metricflow-semantics/metricflow_semantics/naming/object_builder_scheme.py b/metricflow-semantics/metricflow_semantics/naming/object_builder_scheme.py
index b6f11d1ec5..55a3dd51ad 100644
--- a/metricflow-semantics/metricflow_semantics/naming/object_builder_scheme.py
+++ b/metricflow-semantics/metricflow_semantics/naming/object_builder_scheme.py
@@ -38,14 +38,16 @@ def input_str(self, instance_spec: InstanceSpec) -> Optional[str]:
 
     @override
     def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> SpecPattern:
-        if not self.input_str_follows_scheme(input_str):
+        if not self.input_str_follows_scheme(input_str, semantic_manifest_lookup=semantic_manifest_lookup):
             raise ValueError(
                 f"The specified input {repr(input_str)} does not match the input described by the object builder "
                 f"pattern."
             )
         try:
             # TODO: Update when more appropriate parsing libraries are available.
-            call_parameter_sets = PydanticWhereFilter(where_sql_template="{{ " + input_str + " }}").call_parameter_sets
+            call_parameter_sets = PydanticWhereFilter(where_sql_template="{{ " + input_str + " }}").call_parameter_sets(
+                custom_granularity_names=semantic_manifest_lookup.semantic_model_lookup.custom_granularity_names
+            )
         except ParseWhereFilterException as e:
             raise ValueError(f"A spec pattern can't be generated from the input string {repr(input_str)}") from e
 
@@ -121,11 +123,14 @@ def spec_pattern(self, input_str: str, semantic_manifest_lookup: SemanticManifes
         raise RuntimeError("There should have been a return associated with one of the CallParameterSets.")
 
     @override
-    def input_str_follows_scheme(self, input_str: str) -> bool:
+    def input_str_follows_scheme(self, input_str: str, semantic_manifest_lookup: SemanticManifestLookup) -> bool:
         if ObjectBuilderNamingScheme._NAME_REGEX.match(input_str) is None:
             return False
         try:
-            call_parameter_sets = WhereFilterParser.parse_call_parameter_sets("{{ " + input_str + " }}")
+            call_parameter_sets = WhereFilterParser.parse_call_parameter_sets(
+                where_sql_template="{{ " + input_str + " }}",
+                custom_granularity_names=semantic_manifest_lookup.semantic_model_lookup.custom_granularity_names,
+            )
             return_value = (
                 len(call_parameter_sets.dimension_call_parameter_sets)
                 + len(call_parameter_sets.time_dimension_call_parameter_sets)
diff --git a/metricflow-semantics/metricflow_semantics/query/group_by_item/candidate_push_down/push_down_visitor.py b/metricflow-semantics/metricflow_semantics/query/group_by_item/candidate_push_down/push_down_visitor.py
index 479a4c1700..252aba397d 100644
--- a/metricflow-semantics/metricflow_semantics/query/group_by_item/candidate_push_down/push_down_visitor.py
+++ b/metricflow-semantics/metricflow_semantics/query/group_by_item/candidate_push_down/push_down_visitor.py
@@ -12,6 +12,7 @@
 from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
 from typing_extensions import override
 
+from metricflow_semantics.errors.custom_grain_not_supported import error_if_not_standard_grain
 from metricflow_semantics.mf_logging.formatting import indent
 from metricflow_semantics.mf_logging.lazy_formattable import LazyFormat
 from metricflow_semantics.mf_logging.pretty_print import mf_pformat, mf_pformat_many
@@ -401,7 +402,13 @@ def visit_metric_node(self, node: MetricGroupByItemResolutionNode) -> PushDownRe
 
             # If time granularity is not set for the metric, defaults to DAY if available, else the smallest available granularity.
             # Note: ignores any granularity set on input metrics.
-            metric_default_time_granularity = metric_to_use_for_time_granularity_resolution.time_granularity or max(
+            metric_time_granularity: Optional[TimeGranularity] = None
+            if metric_to_use_for_time_granularity_resolution.time_granularity is not None:
+                metric_time_granularity = error_if_not_standard_grain(
+                    context=f"Metric({metric_to_use_for_time_granularity_resolution}).time_granularity",
+                    input_granularity=metric_to_use_for_time_granularity_resolution.time_granularity,
+                )
+            metric_default_time_granularity = metric_time_granularity or max(
                 TimeGranularity.DAY,
                 self._semantic_manifest_lookup.metric_lookup.get_min_queryable_time_granularity(
                     MetricReference(metric_to_use_for_time_granularity_resolution.name)
diff --git a/metricflow-semantics/metricflow_semantics/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py b/metricflow-semantics/metricflow_semantics/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
index 091fddc84a..30b64b0811 100644
--- a/metricflow-semantics/metricflow_semantics/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
+++ b/metricflow-semantics/metricflow_semantics/query/group_by_item/filter_spec_resolution/filter_spec_resolver.py
@@ -331,7 +331,9 @@ def _resolve_specs_for_where_filters(
         for location, where_filters in where_filters_and_locations.items():
             for where_filter in where_filters:
                 try:
-                    filter_call_parameter_sets = where_filter.call_parameter_sets
+                    filter_call_parameter_sets = where_filter.call_parameter_sets(
+                        custom_granularity_names=self._manifest_lookup.semantic_model_lookup.custom_granularity_names
+                    )
                 except Exception as e:
                     non_parsable_resolutions.append(
                         NonParsableFilterResolution(
diff --git a/metricflow-semantics/metricflow_semantics/query/query_parser.py b/metricflow-semantics/metricflow_semantics/query/query_parser.py
index cf0c1f948c..555216e1ab 100644
--- a/metricflow-semantics/metricflow_semantics/query/query_parser.py
+++ b/metricflow-semantics/metricflow_semantics/query/query_parser.py
@@ -210,7 +210,9 @@ def _parse_order_by_names(
                 order_by_name_without_prefix = order_by_name
 
             for group_by_item_naming_scheme in self._group_by_item_naming_schemes:
-                if group_by_item_naming_scheme.input_str_follows_scheme(order_by_name_without_prefix):
+                if group_by_item_naming_scheme.input_str_follows_scheme(
+                    order_by_name_without_prefix, semantic_manifest_lookup=self._manifest_lookup
+                ):
                     possible_inputs.append(
                         ResolverInputForGroupByItem(
                             input_obj=order_by_name,
@@ -223,7 +225,9 @@ def _parse_order_by_names(
                     break
 
             for metric_naming_scheme in self._metric_naming_schemes:
-                if metric_naming_scheme.input_str_follows_scheme(order_by_name_without_prefix):
+                if metric_naming_scheme.input_str_follows_scheme(
+                    order_by_name_without_prefix, semantic_manifest_lookup=self._manifest_lookup
+                ):
                     possible_inputs.append(
                         ResolverInputForMetric(
                             input_obj=order_by_name,
@@ -373,7 +377,9 @@ def _parse_and_validate_query(
         for metric_name in metric_names:
             resolver_input_for_metric: Optional[MetricFlowQueryResolverInput] = None
             for metric_naming_scheme in self._metric_naming_schemes:
-                if metric_naming_scheme.input_str_follows_scheme(metric_name):
+                if metric_naming_scheme.input_str_follows_scheme(
+                    metric_name, semantic_manifest_lookup=self._manifest_lookup
+                ):
                     resolver_input_for_metric = ResolverInputForMetric(
                         input_obj=metric_name,
                         naming_scheme=metric_naming_scheme,
@@ -405,7 +411,9 @@ def _parse_and_validate_query(
         for group_by_name in group_by_names:
             resolver_input_for_group_by_item: Optional[MetricFlowQueryResolverInput] = None
             for group_by_item_naming_scheme in self._group_by_item_naming_schemes:
-                if group_by_item_naming_scheme.input_str_follows_scheme(group_by_name):
+                if group_by_item_naming_scheme.input_str_follows_scheme(
+                    group_by_name, semantic_manifest_lookup=self._manifest_lookup
+                ):
                     spec_pattern = group_by_item_naming_scheme.spec_pattern(
                         group_by_name, semantic_manifest_lookup=self._manifest_lookup
                     )
diff --git a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_dimension.py b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_dimension.py
index 9ca2798a61..409706178c 100644
--- a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_dimension.py
+++ b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_dimension.py
@@ -7,7 +7,6 @@
     DimensionCallParameterSet,
     TimeDimensionCallParameterSet,
 )
-from dbt_semantic_interfaces.naming.dundered import DunderedNameFormatter
 from dbt_semantic_interfaces.protocols.protocol_hint import ProtocolHint
 from dbt_semantic_interfaces.protocols.query_interface import (
     QueryInterfaceDimension,
@@ -18,6 +17,7 @@
 from typing_extensions import override
 
 from metricflow_semantics.errors.error_classes import InvalidQuerySyntax
+from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
 from metricflow_semantics.query.group_by_item.filter_spec_resolution.filter_location import WhereFilterLocation
 from metricflow_semantics.query.group_by_item.filter_spec_resolution.filter_spec_lookup import (
     FilterSpecResolutionLookUp,
@@ -134,15 +134,19 @@ def __init__(  # noqa
         spec_resolution_lookup: FilterSpecResolutionLookUp,
         where_filter_location: WhereFilterLocation,
         rendered_spec_tracker: RenderedSpecTracker,
+        custom_granularity_names: Sequence[str],
     ):
         self._column_association_resolver = column_association_resolver
         self._resolved_spec_lookup = spec_resolution_lookup
         self._where_filter_location = where_filter_location
         self._rendered_spec_tracker = rendered_spec_tracker
+        self._custom_granularity_names = custom_granularity_names
 
     def create(self, name: str, entity_path: Sequence[str] = ()) -> WhereFilterDimension:
         """Create a WhereFilterDimension."""
-        structured_name = DunderedNameFormatter.parse_name(name.lower())
+        structured_name = StructuredLinkableSpecName.from_name(
+            name.lower(), custom_granularity_names=self._custom_granularity_names
+        )
 
         return WhereFilterDimension(
             column_association_resolver=self._column_association_resolver,
diff --git a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_entity.py b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_entity.py
index 080beabfcd..97f95f5439 100644
--- a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_entity.py
+++ b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_entity.py
@@ -5,7 +5,6 @@
 from dbt_semantic_interfaces.call_parameter_sets import (
     EntityCallParameterSet,
 )
-from dbt_semantic_interfaces.naming.dundered import DunderedNameFormatter
 from dbt_semantic_interfaces.protocols.protocol_hint import ProtocolHint
 from dbt_semantic_interfaces.protocols.query_interface import QueryInterfaceEntity, QueryInterfaceEntityFactory
 from dbt_semantic_interfaces.references import EntityReference
@@ -14,6 +13,7 @@
 from typing_extensions import override
 
 from metricflow_semantics.errors.error_classes import InvalidQuerySyntax
+from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
 from metricflow_semantics.query.group_by_item.filter_spec_resolution.filter_location import WhereFilterLocation
 from metricflow_semantics.query.group_by_item.filter_spec_resolution.filter_spec_lookup import (
     FilterSpecResolutionLookUp,
@@ -103,7 +103,7 @@ def __init__(  # noqa
 
     def create(self, entity_name: str, entity_path: Sequence[str] = ()) -> WhereFilterEntity:
         """Create a WhereFilterEntity."""
-        structured_name = DunderedNameFormatter.parse_name(entity_name.lower())
+        structured_name = StructuredLinkableSpecName.from_name(entity_name.lower(), custom_granularity_names=())
 
         return WhereFilterEntity(
             column_association_resolver=self._column_association_resolver,
diff --git a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_transform.py b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_transform.py
index 47ef06476d..7397f7b4d7 100644
--- a/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_transform.py
+++ b/metricflow-semantics/metricflow_semantics/specs/where_filter/where_filter_transform.py
@@ -70,6 +70,7 @@ def create_from_where_filter_intersection(  # noqa: D102
                 spec_resolution_lookup=self._spec_resolution_lookup,
                 where_filter_location=filter_location,
                 rendered_spec_tracker=rendered_spec_tracker,
+                custom_granularity_names=self._semantic_model_lookup.custom_granularity_names,
             )
             time_dimension_factory = WhereFilterTimeDimensionFactory(
                 column_association_resolver=self._column_association_resolver,
diff --git a/metricflow-semantics/tests_metricflow_semantics/naming/test_dunder_naming_scheme.py b/metricflow-semantics/tests_metricflow_semantics/naming/test_dunder_naming_scheme.py
index 2724ead45f..142213125b 100644
--- a/metricflow-semantics/tests_metricflow_semantics/naming/test_dunder_naming_scheme.py
+++ b/metricflow-semantics/tests_metricflow_semantics/naming/test_dunder_naming_scheme.py
@@ -75,13 +75,28 @@ def test_input_str(dunder_naming_scheme: DunderNamingScheme) -> None:  # noqa: D
     )
 
 
-def test_input_follows_scheme(dunder_naming_scheme: DunderNamingScheme) -> None:  # noqa: D103
-    assert dunder_naming_scheme.input_str_follows_scheme("listing__country")
-    assert dunder_naming_scheme.input_str_follows_scheme("listing__creation_time__month")
-    assert dunder_naming_scheme.input_str_follows_scheme("booking__listing")
-    assert not dunder_naming_scheme.input_str_follows_scheme("listing__creation_time__extract_month")
-    assert not dunder_naming_scheme.input_str_follows_scheme("123")
-    assert not dunder_naming_scheme.input_str_follows_scheme("TimeDimension('metric_time')")
+def test_input_follows_scheme(  # noqa: D103
+    dunder_naming_scheme: DunderNamingScheme,
+    simple_semantic_manifest_lookup: SemanticManifestLookup,
+) -> None:
+    assert dunder_naming_scheme.input_str_follows_scheme(
+        "listing__country", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert dunder_naming_scheme.input_str_follows_scheme(
+        "listing__creation_time__month", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert dunder_naming_scheme.input_str_follows_scheme(
+        "booking__listing", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not dunder_naming_scheme.input_str_follows_scheme(
+        "listing__creation_time__extract_month", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not dunder_naming_scheme.input_str_follows_scheme(
+        "123", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not dunder_naming_scheme.input_str_follows_scheme(
+        "TimeDimension('metric_time')", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
 
 
 def test_spec_pattern(  # noqa: D103
diff --git a/metricflow-semantics/tests_metricflow_semantics/naming/test_metric_name_scheme.py b/metricflow-semantics/tests_metricflow_semantics/naming/test_metric_name_scheme.py
index b8e057c491..24c456b5a6 100644
--- a/metricflow-semantics/tests_metricflow_semantics/naming/test_metric_name_scheme.py
+++ b/metricflow-semantics/tests_metricflow_semantics/naming/test_metric_name_scheme.py
@@ -19,8 +19,12 @@ def test_input_str(metric_naming_scheme: MetricNamingScheme) -> None:  # noqa: D
     assert metric_naming_scheme.input_str(MetricSpec(element_name="bookings")) == "bookings"
 
 
-def test_input_follows_scheme(metric_naming_scheme: MetricNamingScheme) -> None:  # noqa: D103
-    assert metric_naming_scheme.input_str_follows_scheme("listings")
+def test_input_follows_scheme(  # noqa: D103
+    metric_naming_scheme: MetricNamingScheme, simple_semantic_manifest_lookup: SemanticManifestLookup
+) -> None:
+    assert metric_naming_scheme.input_str_follows_scheme(
+        "listings", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
 
 
 def test_spec_pattern(  # noqa: D103
diff --git a/metricflow-semantics/tests_metricflow_semantics/naming/test_object_builder_naming_scheme.py b/metricflow-semantics/tests_metricflow_semantics/naming/test_object_builder_naming_scheme.py
index 1b8a058a32..5f471d0c47 100644
--- a/metricflow-semantics/tests_metricflow_semantics/naming/test_object_builder_naming_scheme.py
+++ b/metricflow-semantics/tests_metricflow_semantics/naming/test_object_builder_naming_scheme.py
@@ -64,20 +64,30 @@ def test_input_str(object_builder_naming_scheme: ObjectBuilderNamingScheme) -> N
     )
 
 
-def test_input_follows_scheme(object_builder_naming_scheme: ObjectBuilderNamingScheme) -> None:  # noqa: D103
+def test_input_follows_scheme(  # noqa: D103
+    object_builder_naming_scheme: ObjectBuilderNamingScheme, simple_semantic_manifest_lookup: SemanticManifestLookup
+) -> None:
     assert object_builder_naming_scheme.input_str_follows_scheme(
-        "Dimension('listing__country', entity_path=['booking'])"
+        "Dimension('listing__country', entity_path=['booking'])",
+        semantic_manifest_lookup=simple_semantic_manifest_lookup,
     )
     assert object_builder_naming_scheme.input_str_follows_scheme(
         "TimeDimension('listing__creation_time', time_granularity_name='month', date_part_name='day', "
-        "entity_path=['booking'])"
+        "entity_path=['booking'])",
+        semantic_manifest_lookup=simple_semantic_manifest_lookup,
     )
     assert object_builder_naming_scheme.input_str_follows_scheme(
-        "Entity('user', entity_path=['booking', 'listing'])",
+        "Entity('user', entity_path=['booking', 'listing'])", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not object_builder_naming_scheme.input_str_follows_scheme(
+        "listing__creation_time__extract_month", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not object_builder_naming_scheme.input_str_follows_scheme(
+        "123", semantic_manifest_lookup=simple_semantic_manifest_lookup
+    )
+    assert not object_builder_naming_scheme.input_str_follows_scheme(
+        "NotADimension('listing__country')", semantic_manifest_lookup=simple_semantic_manifest_lookup
     )
-    assert not object_builder_naming_scheme.input_str_follows_scheme("listing__creation_time__extract_month")
-    assert not object_builder_naming_scheme.input_str_follows_scheme("123")
-    assert not object_builder_naming_scheme.input_str_follows_scheme("NotADimension('listing__country')")
 
 
 def test_spec_pattern(  # noqa: D103
diff --git a/metricflow/dataflow/builder/dataflow_plan_builder.py b/metricflow/dataflow/builder/dataflow_plan_builder.py
index d41d22d047..c53fe36236 100644
--- a/metricflow/dataflow/builder/dataflow_plan_builder.py
+++ b/metricflow/dataflow/builder/dataflow_plan_builder.py
@@ -19,6 +19,7 @@
 from dbt_semantic_interfaces.validations.unique_valid_name import MetricFlowReservedKeywords
 from metricflow_semantics.dag.id_prefix import StaticIdPrefix
 from metricflow_semantics.dag.mf_dag import DagId
+from metricflow_semantics.errors.custom_grain_not_supported import error_if_not_standard_grain
 from metricflow_semantics.errors.error_classes import UnableToSatisfyQueryError
 from metricflow_semantics.filters.time_constraint import TimeRangeConstraint
 from metricflow_semantics.mf_logging.formatting import indent
@@ -512,6 +513,16 @@ def _build_base_metric_output_node(
             len(metric.input_measures) == 1
         ), f"A base metric should not have multiple measures. Got {metric.input_measures}"
 
+        cumulative_grain_to_date: Optional[TimeGranularity] = None
+        if (
+            metric.type_params.cumulative_type_params
+            and metric.type_params.cumulative_type_params.grain_to_date is not None
+        ):
+            cumulative_grain_to_date = error_if_not_standard_grain(
+                context=f"CumulativeMetric({metric_spec.element_name}).grain_to_date",
+                input_granularity=metric.type_params.cumulative_type_params.grain_to_date,
+            )
+
         metric_input_measure_spec = self._build_input_measure_spec(
             filter_spec_factory=filter_spec_factory,
             metric=metric,
@@ -526,11 +537,7 @@ def _build_base_metric_output_node(
                         if metric.type_params.cumulative_type_params
                         else None
                     ),
-                    cumulative_grain_to_date=(
-                        metric.type_params.cumulative_type_params.grain_to_date
-                        if metric.type_params.cumulative_type_params
-                        else None
-                    ),
+                    cumulative_grain_to_date=cumulative_grain_to_date,
                 )
                 if metric.type is MetricType.CUMULATIVE
                 else None
@@ -1399,6 +1406,13 @@ def _build_input_metric_specs_for_derived_metric(
                 ),
             )
 
+            input_metric_offset_to_grain: Optional[TimeGranularity] = None
+            if input_metric.offset_to_grain is not None:
+                input_metric_offset_to_grain = error_if_not_standard_grain(
+                    context=f"Metric({metric.name}).InputMetric({input_metric.name}).offset_to_grain",
+                    input_granularity=input_metric.offset_to_grain,
+                )
+
             spec = MetricSpec(
                 element_name=input_metric.name,
                 filter_spec_set=filter_spec_set,
@@ -1411,7 +1425,7 @@ def _build_input_metric_specs_for_derived_metric(
                     if input_metric.offset_window
                     else None
                 ),
-                offset_to_grain=input_metric.offset_to_grain,
+                offset_to_grain=input_metric_offset_to_grain,
             )
             input_metric_specs.append(spec)
         return tuple(input_metric_specs)
@@ -1514,7 +1528,9 @@ def _build_aggregated_measure_from_measure_source_node(
             granularity: Optional[TimeGranularity] = None
             count = 0
             if cumulative_window is not None:
-                granularity = cumulative_window.granularity
+                granularity = error_if_not_standard_grain(
+                    context="CumulativeMetric.window.granularity", input_granularity=cumulative_window.granularity
+                )
                 count = cumulative_window.count
             elif cumulative_grain_to_date is not None:
                 count = 1
diff --git a/metricflow/dataflow/nodes/join_conversion_events.py b/metricflow/dataflow/nodes/join_conversion_events.py
index cc6530996b..5bdaaee2ec 100644
--- a/metricflow/dataflow/nodes/join_conversion_events.py
+++ b/metricflow/dataflow/nodes/join_conversion_events.py
@@ -77,7 +77,7 @@ def accept(self, visitor: DataflowPlanNodeVisitor[VisitorOutputT]) -> VisitorOut
 
     @property
     def description(self) -> str:  # noqa: D102
-        return f"Find conversions for {self.entity_spec.qualified_name} within the range of {f'{self.window.count} {self.window.granularity.value}' if self.window else 'INF'}"
+        return f"Find conversions for {self.entity_spec.qualified_name} within the range of {f'{self.window.count} {self.window.granularity}' if self.window else 'INF'}"
 
     @property
     def displayed_properties(self) -> Sequence[DisplayedProperty]:  # noqa: D102
diff --git a/metricflow/plan_conversion/sql_join_builder.py b/metricflow/plan_conversion/sql_join_builder.py
index 13954157f7..8aa8598237 100644
--- a/metricflow/plan_conversion/sql_join_builder.py
+++ b/metricflow/plan_conversion/sql_join_builder.py
@@ -6,6 +6,7 @@
 from dbt_semantic_interfaces.protocols.metric import MetricTimeWindow
 from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
 from metricflow_semantics.assert_one_arg import assert_exactly_one_arg_set
+from metricflow_semantics.errors.custom_grain_not_supported import error_if_not_standard_grain
 from metricflow_semantics.sql.sql_join_type import SqlJoinType
 
 from metricflow.dataflow.nodes.join_conversion_events import JoinConversionEventsNode
@@ -466,7 +467,9 @@ def _make_time_range_window_join_condition(
                 right_expr=SqlSubtractTimeIntervalExpression.create(
                     arg=time_comparison_column_expr,
                     count=window.count,
-                    granularity=window.granularity,
+                    granularity=error_if_not_standard_grain(
+                        input_granularity=window.granularity,
+                    ),
                 ),
             )
             comparison_expressions.append(start_of_range_comparison_expr)
@@ -551,7 +554,9 @@ def make_join_to_time_spine_join_description(
         )
         if node.offset_window:
             left_expr = SqlSubtractTimeIntervalExpression.create(
-                arg=left_expr, count=node.offset_window.count, granularity=node.offset_window.granularity
+                arg=left_expr,
+                count=node.offset_window.count,
+                granularity=error_if_not_standard_grain(input_granularity=node.offset_window.granularity),
             )
         elif node.offset_to_grain:
             left_expr = SqlDateTruncExpression.create(time_granularity=node.offset_to_grain, arg=left_expr)