Skip to content
Merged
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
25 changes: 25 additions & 0 deletions .script/tests/asimParsersTest/ingestTimeParser/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ discover target
→ freeze a typed parser-input snapshot from the fixture
→ evaluate ParserFilterQuery against an inline fixture datatable
→ upload fixture rows directly through the parser DCR
→ verify exact per-timestamp target visibility
→ run ASIM data checks
→ optionally replay the fixture to SourceRawTable using InputStream
→ execute the mapped QTP and compare parity when replay succeeds
Expand All @@ -62,6 +63,8 @@ optional. If its table or DCR cannot be established, the report records a
warning and blocks QTP/parity without failing the required parser validation
stages. Once the raw DCR exists, upload or visibility failures fail the target;
QTP execution and parity also fail the target when they run and fail.
Parser ingestion is reported as passed only when both upload and target
visibility pass.

Static validation requires `ParserFilterQuery` to contain a `where` stage when
`ParserQuery` contains one. This is a presence check only; it does not require
Expand All @@ -76,6 +79,28 @@ must use the supported stream-name format.
available in `validation-result.json` when visibility detects unexpected,
duplicate, or incomplete rows.

Exact-row visibility timeouts include `expectedCounts`, the last successful
`observedCounts`, `missingTimestamps` (timestamps with fewer rows than expected),
and `observedQueryTable` in the error details, including when the shared deadline
expires between polls or during settling. Before any successful observation,
the observed counts are empty and the query table is null. Shared-deadline
errors retain their original stage and message; query failures remain unchanged.
The validation artifact records `targetVisibility` as `passed`, `failed`, or
`not-run`; failed outcomes retain their original error stage and diagnostic
details. The report nests this explicit result under the existing
`parserIngestion` stage, whose public key remains unchanged. Older artifacts
without `targetVisibility` use the exact legacy `target-visibility` error stage
or a generic deadline error together with target-visibility timing and no
output-validation timing to identify this failure; other generic query errors
are not attributed to visibility.

The report's `schema` stage reflects the actual
`outputValidation.schemaValidation` result and is `not-run` when that evidence
is absent, even if a parser DCR exists. That field is emitted only after
ASimDataTester succeeds and its authority is `parser-dcr-output-stream`; it
does not represent a separate ASimSchemaTester execution. These diagnostics do
not resolve the underlying canary ingestion timeout.

QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and
`itp_rows` under `outputValidation`. It derives `intersect_schema` from shared
column names, excluding the platform-managed `Type` column, and records both
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -497,69 +497,96 @@ def wait_for_exact_rows(
raise ValueError("Expected counts cannot be negative.")
poll_index = 0
stable_once = False
while True:
query_table = self.query_exact_rows(table_name, timestamps)
raw_observed_counts = _query_timestamp_counts(
query_table, stage
)
unexpected = {
timestamp: count
for timestamp, count in raw_observed_counts.items()
if timestamp not in expected_counts
}
if unexpected:
raise HarnessError(
stage,
"Exact timestamp query returned unexpected timestamps.",
details={
"unexpected": unexpected,
"observedQueryTable": query_table,
},
query_table = None
observed_counts: dict[str, int] = {}
try:
while True:
query_table = self.query_exact_rows(table_name, timestamps)
raw_observed_counts = _query_timestamp_counts(
query_table, stage
)
observed_counts = {
timestamp: raw_observed_counts.get(timestamp, 0)
for timestamp in expected_counts
}
overages = {
timestamp: {
"expected": expected_counts.get(timestamp, 0),
"observed": observed,
unexpected = {
timestamp: count
for timestamp, count in raw_observed_counts.items()
if timestamp not in expected_counts
}
if unexpected:
raise HarnessError(
stage,
"Exact timestamp query returned unexpected timestamps.",
details={
"unexpected": unexpected,
"observedQueryTable": query_table,
},
)
observed_counts = {
timestamp: raw_observed_counts.get(timestamp, 0)
for timestamp in expected_counts
}
for timestamp, observed in observed_counts.items()
if observed > expected_counts.get(timestamp, 0)
overages = {
timestamp: {
"expected": expected_counts.get(timestamp, 0),
"observed": observed,
}
for timestamp, observed in observed_counts.items()
if observed > expected_counts.get(timestamp, 0)
}
if overages:
raise HarnessError(
stage,
"Exact timestamp slice contains more rows than "
"expected for one or more timestamps.",
details={
"overages": overages,
"observedQueryTable": query_table,
},
)
if observed_counts == expected_counts:
if stable_once:
return query_table
stable_once = True
self._sleep(self.settling_delay_seconds)
continue
stable_once = False
if self._remaining() <= 0:
raise HarnessError(
stage,
"Timed out waiting for exact table rows.",
details={
"expectedCounts": expected_counts,
"observedCounts": observed_counts,
"missingTimestamps": [
timestamp
for timestamp, count in expected_counts.items()
if observed_counts.get(timestamp, 0) < count
],
"observedQueryTable": query_table,
},
)
delay = self.poll_delays[
min(poll_index, len(self.poll_delays) - 1)
]
poll_index += 1
self._sleep(delay)
except HarnessError as exc:
# Only enrich locally generated deadline errors, not query failures.
if (
exc.stage not in {"query", "deadline"}
or str(exc) != "Validation deadline expired."
or exc.status_code is not None
):
raise
exc.details = {
"expectedCounts": expected_counts,
"observedCounts": observed_counts,
"missingTimestamps": [
timestamp
for timestamp, count in expected_counts.items()
if observed_counts.get(timestamp, 0) < count
],
"observedQueryTable": query_table,
}
if overages:
raise HarnessError(
stage,
"Exact timestamp slice contains more rows than "
"expected for one or more timestamps.",
details={
"overages": overages,
"observedQueryTable": query_table,
},
)
if observed_counts == expected_counts:
if stable_once:
return query_table
stable_once = True
self._sleep(self.settling_delay_seconds)
continue
stable_once = False
if self._remaining() <= 0:
raise HarnessError(
stage,
"Timed out waiting for exact table rows.",
details={
"expectedCounts": expected_counts,
"observedCounts": observed_counts,
"observedQueryTable": query_table,
},
)
delay = self.poll_delays[
min(poll_index, len(self.poll_delays) - 1)
]
poll_index += 1
self._sleep(delay)
raise

def upload_rows(
self,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
HARNESS_RESULT_VERSION,
HarnessError,
ValidationRunner,
_query_timestamp_counts,
_validate_timestamp,
_write_json_no_clobber,
build_fixture_snapshot,
Expand All @@ -20,9 +21,17 @@
)


def _error_payload(exc: HarnessError) -> dict[str, Any]:
def _error_payload(
exc: HarnessError,
*,
include_observed_query_table: bool = False,
) -> dict[str, Any]:
details = exc.details
if isinstance(details, Mapping) and "observedQueryTable" in details:
if (
not include_observed_query_table
and isinstance(details, Mapping)
and "observedQueryTable" in details
):
details = dict(details)
details.pop("observedQueryTable", None)
return {
Expand Down Expand Up @@ -173,6 +182,7 @@ def run(
"parserDcr": None,
"rawUpload": [],
"parserUpload": [],
"targetVisibility": {"status": "not-run"},
"snapshotPath": str(snapshot_output),
"filterValidation": None,
"outputValidation": None,
Expand Down Expand Up @@ -273,13 +283,36 @@ def run(
reconcile_table=target_table,
expected_counts_by_timestamp=normalized_target_counts,
)
with self._stage("target-visibility"):
self.wait_for_exact_rows(
target_table,
list(normalized_target_counts),
normalized_target_counts,
stage="target-visibility",
)
try:
with self._stage("target-visibility"):
observed_query_table = self.wait_for_exact_rows(
target_table,
list(normalized_target_counts),
normalized_target_counts,
stage="target-visibility",
)
except HarnessError as exc:
result["targetVisibility"] = {
"status": "failed",
"error": _error_payload(
exc,
include_observed_query_table=True,
),
}
Comment on lines +294 to +301
raise
observed_counts = _query_timestamp_counts(
observed_query_table,
"target-visibility",
)
result["targetVisibility"] = {
"status": "passed",
"expectedCounts": normalized_target_counts,
"observedCounts": {
timestamp: observed_counts.get(timestamp, 0)
for timestamp in normalized_target_counts
},
"missingTimestamps": [],
}

if validation_runner is not None:
with self._stage("output-validation"):
Expand Down
Loading
Loading