Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 28 additions & 3 deletions src/winml/modelkit/analyze/core/runtime_checker_query.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import onnx
import pandas as pd
from onnx import numpy_helper
from onnx.defs import SchemaError

from ...onnx import (
ONNXDomain,
Expand Down Expand Up @@ -590,7 +591,18 @@ def get_query_conditions_for_node(
- is_qdq: True if node has QDQ quantization on inputs or outputs.
"""
conditions = {}
schema = domain.get_op_schema(node.op_type, opset_version)
try:
schema = domain.get_op_schema(node.op_type, opset_version)
except SchemaError as e:
# Some runtime-specific models emit extension ops in the default ONNX
# domain where no standard schema exists (for example
# SimplifiedLayerNormalization). Treat this as unsupported instead of
# aborting the whole analyze command.
raise OpUnsupportedError(
Comment thread
fangyangci marked this conversation as resolved.
"Node "
f"{node.op_type} has no registered schema for domain "
f"'{domain.schema_domain}' at opset {opset_version}: {e}"
) from e
input_names, variadic_input_name, attribute_names, type_annotations = get_op_input_properties(
schema
)
Expand Down Expand Up @@ -2685,7 +2697,20 @@ def get_pattern_id(is_qdq: bool) -> str:
) as e:
conditions_ms = _elapsed_ms(conditions_start)
exception_type = type(e).__name__
logger.error(
reason = "optional_input_properties_not_found"
log_fn = logger.error

if isinstance(e, OpUnsupportedError):
error_message = str(e)
if "has no registered schema for domain" in error_message:
reason = f"schema_not_registered:{node.op_type}"
else:
reason = f"unsupported_op:{node.op_type}"
log_fn = logger.debug
elif isinstance(e, OpLackOfRequiredInformationError):
reason = "required_information_missing"

log_fn(
"%s caught for op %s (node: %s): %s",
exception_type,
node.op_type,
Expand All @@ -2710,7 +2735,7 @@ def get_pattern_id(is_qdq: bool) -> str:
compile=False,
run=False,
no_data=True,
reason="optional_input_properties_not_found",
reason=reason,
node_tags=node_tags,
debug_details=conditions_error_debug_details,
),
Expand Down
89 changes: 88 additions & 1 deletion tests/unit/analyze/core/test_runtime_checker_query_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
node_to_pattern_match,
try_load_external_initializer_array,
)
from winml.modelkit.analyze.exceptions import OpOptionalInputSupportError
from winml.modelkit.analyze.exceptions import OpOptionalInputSupportError, OpUnsupportedError
from winml.modelkit.analyze.utils.model_utils import DUMMY_FLOAT
from winml.modelkit.analyze.utils.node_key_utils import resolve_stable_node_key
from winml.modelkit.onnx import ONNXDomain
Expand Down Expand Up @@ -129,6 +129,29 @@ def test_preserves_column_order_for_present_entries(self):
class TestGetQueryConditionsForNode:
"""Test condition extraction for runtime rule lookups."""

def test_missing_schema_is_reported_as_unsupported_error(self):
"""Unknown schema should be classified as unsupported instead of crashing analyze."""
node = helper.make_node(
"SimplifiedLayerNormalization",
["X", "gamma"],
["Y"],
name="rms_norm",
)

with pytest.raises(OpUnsupportedError) as exc_info:
get_query_conditions_for_node(
node=node,
opset_version=21,
valueinfo={},
initializers={},
constants={},
domain=ONNXDomain.AI_ONNX,
input_to_dq={},
output_to_q={},
)

assert "has no registered schema" in str(exc_info.value)

def test_external_initializer_without_payload_is_not_marked_constant(self):
"""External-data initializers without loaded values keep shape but not constant status."""
node = helper.make_node("Add", ["weight", "input"], ["output"], name="add_node")
Expand Down Expand Up @@ -258,6 +281,70 @@ def test_try_load_external_initializer_array_returns_plain_ndarray(
assert renamed_sidecar_path.exists()


class TestRunForNodeUnsupportedReasons:
"""End-to-end reason mapping tests for RuntimeCheckerQuery.run_for_node."""

@staticmethod
def _build_identity_model() -> onnx.ModelProto:
input_info = helper.make_tensor_value_info("input", TensorProto.FLOAT, [1])
output_info = helper.make_tensor_value_info("output", TensorProto.FLOAT, [1])
node = helper.make_node("Identity", ["input"], ["output"], name="identity_node")
graph = helper.make_graph([node], "identity_graph", [input_info], [output_info])
return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 21)])

def test_run_for_node_preserves_schema_missing_reason(self, monkeypatch):
"""Schema lookup misses should surface schema-specific reason in returned result."""
model = self._build_identity_model()
node = model.graph.node[0]
query = RuntimeCheckerQuery(model, ep_name="QNNExecutionProvider", device_type="NPU")
query.node_checkers = []

def _raise_schema_miss(*args, **kwargs):
del args, kwargs
raise OpUnsupportedError(
"Node Identity has no registered schema for domain '' at opset 21"
)

monkeypatch.setattr(
runtime_checker_query_module,
"get_query_conditions_for_node",
_raise_schema_miss,
)

result = query.run_for_node(node, for_debug=False, run_unknown_op=False)

assert result.result.no_data is True
assert result.result.compile is False
assert result.result.run is False
assert result.result.reason == "schema_not_registered:Identity"
assert result.result.debug_details is None

def test_run_for_node_preserves_generic_unsupported_reason(self, monkeypatch):
"""Unsupported-op path should keep a specific unsupported reason in result payload."""
model = self._build_identity_model()
node = model.graph.node[0]
query = RuntimeCheckerQuery(model, ep_name="QNNExecutionProvider", device_type="NPU")
query.node_checkers = []

def _raise_generic_unsupported(*args, **kwargs):
del args, kwargs
raise OpUnsupportedError("Node Identity is not supported")

monkeypatch.setattr(
runtime_checker_query_module,
"get_query_conditions_for_node",
_raise_generic_unsupported,
)

result = query.run_for_node(node, for_debug=False, run_unknown_op=False)

assert result.result.no_data is True
assert result.result.compile is False
assert result.result.run is False
assert result.result.reason == "unsupported_op:Identity"
assert result.result.debug_details is None


class TestLocalEPFallback:
"""Test local EP fallback helpers for single-node execution."""

Expand Down
Loading