[ASIM] Decouple parser validation from optional QTP replay

.script/tests/asimParsersTest/ingestTimeParser/README.md

@@ -48,15 +48,27 @@ asim_ingestion/
 discover target
   → static parser validation
   → prepare immutable fixture
-  → raw pass-through ingestion
-  → freeze raw-table source snapshot
-  → evaluate ParserFilterQuery
-  → execute ParserQuery through a real DCR
+  → freeze a typed parser-input snapshot from the fixture
+  → evaluate ParserFilterQuery against an inline fixture datatable
+  → upload fixture rows directly through the parser DCR
   → run ASIM data checks
-  → optionally execute the mapped QTP and compare parity
+  → optionally replay the fixture to SourceRawTable using InputStream
+  → execute the mapped QTP and compare parity when replay succeeds
   → clean up DCRs and publish the report
 ```
 
+Parser validation does not depend on raw-table replay. QTP source replay is
+optional: if its table, DCR, upload, or visibility stage fails, the report
+records a warning and blocks QTP/parity without failing the required parser
+validation stages.
+
+QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and
+`itp_rows` under `outputValidation`. It derives `intersect_schema` from shared
+column names and records both QTP and ITP types. A shared-column type mismatch
+fails schema parity. When shared types are consistent, it filters both row
+sets into `qtp_rows_filtered` and `itp_rows_filtered` and runs the typed
+duplicate-aware row comparison across that shared schema.
+
 ## Tests
 
 Run the complete regression suite from the repository root:

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py

@@ -97,6 +97,15 @@ def validate_outputs(
     ) -> Mapping[str, Any]:
         raise NotImplementedError
 
+    def validate_parity(
+        self,
+        query_client: Any,
+        source_snapshot: Mapping[str, Any],
+        parser_document: Mapping[str, Any],
+        query_time_document: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        raise NotImplementedError
+
 
 class DefaultAzureTokenProvider:
     def __init__(self):
@@ -253,97 +262,105 @@ def build_exact_timestamp_query(
     )
 
 
-def _normalize_query_columns(
-    columns: Sequence[Mapping[str, Any]],
-) -> list[dict[str, str]]:
-    schema: list[dict[str, str]] = []
-    seen: set[str] = set()
-    for column in columns:
-        name = column.get("name")
-        column_type = column.get("type")
-        if not isinstance(name, str) or not isinstance(column_type, str):
-            raise HarnessError(
-                "query",
-                "Log Analytics returned an invalid column definition.",
-                details=column,
-            )
-        if name in seen:
-            raise HarnessError(
-                "query", f"Log Analytics returned duplicate column {name}."
-            )
-        seen.add(name)
-        schema.append(
-            {"name": name, "type": _normalize_dcr_type(column_type)}
-        )
-    return schema
-
-
-def build_source_snapshot(
+def build_fixture_snapshot(
     *,
     target_id: str,
     table_name: str,
     timestamps: Sequence[str],
-    query_table: Mapping[str, Any],
+    schema: Sequence[Mapping[str, str]],
+    fixture_rows: Sequence[Mapping[str, Any]],
 ) -> dict[str, Any]:
-    columns = query_table.get("columns")
-    raw_rows = query_table.get("rows")
-    if not isinstance(columns, list) or not isinstance(raw_rows, list):
-        raise HarnessError(
-            "source-snapshot",
-            "Log Analytics query response must contain columns and rows.",
+    normalized_schema: list[dict[str, str]] = [
+        {
+            "name": str(column["name"]),
+            "type": _normalize_dcr_type(str(column["type"])),
+        }
+        for column in schema
+    ]
+    known_names = {
+        column["name"] for column in normalized_schema
+    }
+    extra_names = [
+        str(name)
+        for row in fixture_rows
+        for name in row
+        if str(name) not in known_names
+        and str(name) not in REPLAY_EXCLUDED_COLUMNS
+    ]
+    for name in dict.fromkeys(extra_names):
+        values = [
+            row.get(name)
+            for row in fixture_rows
+            if row.get(name) is not None
+        ]
+        if values and all(isinstance(value, bool) for value in values):
+            column_type = "boolean"
+        elif values and all(
+            isinstance(value, int) and not isinstance(value, bool)
+            for value in values
+        ):
+            column_type = "long"
+        elif values and all(
+            isinstance(value, (int, float))
+            and not isinstance(value, bool)
+            for value in values
+        ):
+            column_type = "real"
+        elif values and all(
+            isinstance(value, (dict, list))
+            for value in values
+        ):
+            column_type = "dynamic"
+        else:
+            column_type = "string"
+        normalized_schema.append(
+            {"name": name, "type": column_type}
         )
-    query_schema = _normalize_query_columns(columns)
-    column_names = [column["name"] for column in query_schema]
-    if any(not isinstance(row, list) for row in raw_rows):
+        known_names.add(name)
+    column_names = [column["name"] for column in normalized_schema]
+    if len(column_names) != len(set(column_names)):
         raise HarnessError(
-            "source-snapshot", "Query rows must be arrays."
+            "source-snapshot",
+            "Fixture schema contains duplicate columns.",
         )
-
-    included_schema = [
-        column
-        for column in query_schema
-        if column["name"] not in REPLAY_EXCLUDED_COLUMNS
-    ]
-    included_names = [column["name"] for column in included_schema]
     type_by_name = {
-        column["name"]: column["type"] for column in included_schema
+        column["name"]: column["type"]
+        for column in normalized_schema
     }
     timestamp_set = {_validate_timestamp(value) for value in timestamps}
-    rows: list[dict[str, Any]] = []
-    for row in raw_rows:
-        if len(row) != len(column_names):
-            raise HarnessError(
-                "source-snapshot",
-                "Query row length does not match the returned schema.",
-            )
-        source = dict(zip(column_names, row))
-        prepared = {
-            name: _json_safe_value(source.get(name), type_by_name[name])
-            for name in included_names
+    rows = [
+        {
+            name: _json_safe_value(row.get(name), type_by_name[name])
+            for name in column_names
         }
-        timestamp = prepared.get("TimeGenerated")
+        for row in fixture_rows
+    ]
+    for row in rows:
+        timestamp = row.get("TimeGenerated")
         if not isinstance(timestamp, str) or timestamp not in timestamp_set:
             raise HarnessError(
                 "source-snapshot",
-                "Query returned a row outside the exact timestamp set.",
+                "Fixture row is outside the exact timestamp set.",
                 details={"TimeGenerated": timestamp},
             )
-        rows.append(prepared)
-
     timestamp_hash = hashlib.sha256(
         "\n".join(sorted(timestamp_set)).encode("utf-8")
     ).hexdigest()
+    fixture_columns = {
+        str(name)
+        for row in fixture_rows
+        for name in row
+    }
     return {
         "version": SNAPSHOT_VERSION,
         "targetId": target_id,
         "sourceRawTable": table_name,
+        "authority": "prepared-fixture",
         "rowCount": len(rows),
         "timestampValues": sorted(timestamp_set),
         "timestampSetHash": f"sha256:{timestamp_hash}",
-        "excludedColumns": sorted(
-            set(column_names) & REPLAY_EXCLUDED_COLUMNS
-        ),
-        "schema": included_schema,
+        "excludedColumns": sorted(fixture_columns - set(column_names)),
+        "schema": normalized_schema,
         "rows": rows,
     }