"""Compare monthly_cloud_billing_revrec outputs from BigQuery inputs and Databend inputs. This output-level parity harness was introduced for issue #35 and broadened into the #48 comparable matrix. It runs the ported Python revrec logic for the same billing month, order snapshot date, and revenue mode over: - legacy BigQuery inputs in gcp-digital-bi.dwh_finance - datalake inputs in transform_dbt_dwh The harness compares the resulting collections and orders outputs before writing, so it can check recognised and predicted modes without clobbering the production output tables. Usage: uv run --extra pyjobs --env-file .env -- \ python transform/python/parity/monthly_cloud_billing_revrec.py \ --billing-month 2026-04 --order-date 2026-05-21 --revenue all uv run --extra pyjobs --env-file .env -- \ python transform/python/parity/monthly_cloud_billing_revrec.py \ --billing-month all --order-date latest --revenue all """ from __future__ import annotations import argparse import json import logging import math import subprocess import sys from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any import pandas as pd PYTHON_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PYTHON_ROOT)) from engine import databend_io # noqa: E402 from jobs import monthly_cloud_billing_revrec as job # noqa: E402 DEFAULT_BIGQUERY_PROJECT = "gcp-digital-bi" DEFAULT_BIGQUERY_DATASET = "dwh_finance" DEFAULT_DATABEND_INPUT_DB = "transform_dbt_dwh" COLLECTION_KEY = ("bill_month", "adjust_tenant_id") ORDER_KEY = ("tenant_id", "bill_month", "order_number") DATE_COLUMNS = { "date_day", "billing_start_date", "billing_end_date", "segment_a_start_date", "segment_a_end_date", "segment_b_start_date", "segment_b_end_date", "segment_c_start_date", "segment_c_end_date", "order_start_date", "order_end_date", "write_off_date", "order_service_start_date", "order_service_end_date", "order_termination_date", "real_order_termination_date", "order_billing_start_date", "order_billing_end_date", "start_date", "end_date", } BOOL_COLUMNS = { "is_salesforce_customer", "order_is_terminated", "is_commitment_plan", "is_new_order", "is_new_salesforce_order", "is_order_closing", } ORDER_NUMERIC = { "total_fees", "order_net_fees", "accumulated_revenue", "accumulated_revenue_before_adjust", "accumulated_credit_amount", "tax_rate", "opportunity_exchange_rate", } BASE_OUTPUT_NUMERIC = { "marketplace_amount", "wire_transfer_amount", "total_amount", "invoice_amount", "order_deferred_revenue_usd", "order_total_fees_usd", "order_accumulated_revenue_before_adjust", "order_accumulated_credit_amount", "revenue_amount", "credit_amount", "adjusted_revenue_amount", "adjusted_shortfall_amount", "billing_allocation_amount", "end_of_contract_shortfall_amount", } OUTPUT_NUMERIC = ( BASE_OUTPUT_NUMERIC | {f"{category}_revenue_amount" for category in job.REVENUE_CATEGORIES} | {f"adjusted_{category}_amount" for category in job.REVENUE_CATEGORIES} ) @dataclass class FieldSummary: matches: int = 0 mismatches: int = 0 examples: list[dict[str, Any]] | None = None def add_example(self, key: tuple[str, ...], bq: Any, databend: Any, limit: int) -> None: if self.examples is None: self.examples = [] if len(self.examples) < limit: self.examples.append({"key": list(key), "bigquery": bq, "databend": databend}) @dataclass(frozen=True) class CaseSpec: billing_month: str order_date: str def _date_str(value: Any) -> str: if isinstance(value, datetime): return value.date().isoformat() if isinstance(value, date): return value.isoformat() return str(value)[:10] def _databend_cursor(): return databend_io.connect() def bq_json(query: str, project: str, *, max_rows: int) -> list[dict[str, Any]]: result = subprocess.run( [ "bq", "query", f"--project_id={project}", "--nouse_legacy_sql", "--format=json", f"--max_rows={max_rows}", query, ], check=True, text=True, capture_output=True, ) return json.loads(result.stdout) def count_bigquery_query(project: str, query: str) -> int: rows = bq_json( f"SELECT COUNT(*) AS row_count FROM ({query}) AS parity_source", project, max_rows=1, ) return int(rows[0]["row_count"]) def fetch_bigquery_query( project: str, query: str, *, count_query: str | None = None, label: str = "BigQuery query", ) -> list[dict[str, Any]]: expected_rows = count_bigquery_query(project, count_query or query) if expected_rows == 0: return [] rows = bq_json(query, project, max_rows=expected_rows + 1) if len(rows) != expected_rows: raise RuntimeError( f"{label}: BigQuery returned {len(rows)} rows, expected {expected_rows}; parity fetch was truncated" ) return rows def quote_sql_values(values: list[str]) -> str: return ", ".join("'" + value.replace("'", "''") + "'" for value in values) def fetch_bigquery_values(project: str, dataset: str, table: str, column: str) -> list[str]: value_query = f""" SELECT DISTINCT CAST({column} AS STRING) AS value FROM `{project}.{dataset}.{table}` WHERE {column} IS NOT NULL """ rows = fetch_bigquery_query( project, f""" SELECT DISTINCT CAST({column} AS STRING) AS value FROM `{project}.{dataset}.{table}` WHERE {column} IS NOT NULL ORDER BY value """, count_query=value_query, label=f"{table}.{column} distinct values", ) return [str(row["value"]) for row in rows] def fetch_databend_values(input_db: str, table: str, column: str) -> list[str]: cur = _databend_cursor() cur.execute( f""" SELECT DISTINCT {column} AS value FROM {input_db}.{table} WHERE {column} IS NOT NULL ORDER BY value """ ) values = [] for row in cur.fetchall(): row_values = list(row.values()) if hasattr(row, "values") else list(row) values.append(_date_str(row_values[0])) return values def common_values(bigquery_values: list[str], databend_values: list[str]) -> list[str]: return sorted(set(bigquery_values) & set(databend_values)) def value_gaps(left_values: list[str], right_values: list[str]) -> list[str]: return sorted(set(left_values) - set(right_values)) def resolve_selector(selector: str, available: list[str], label: str) -> list[str]: if not available: raise RuntimeError(f"No common {label} values are available for parity") if selector == "latest": return [available[-1]] if selector == "all": return list(available) if selector not in available: raise RuntimeError( f"Requested {label} {selector} is not present in both BigQuery and Databend; " f"available range is {available[0]}..{available[-1]}" ) return [selector] def discover_common_cases(args: argparse.Namespace) -> tuple[list[CaseSpec], dict[str, Any]]: bq_billing_months = fetch_bigquery_values( args.bigquery_project, args.bigquery_dataset, "salesforce_org_monthly_revrec", "billing_month", ) db_billing_months = fetch_databend_values( args.databend_input_db, "salesforce_org_monthly_revrec", "billing_month", ) bq_order_dates = fetch_bigquery_values( args.bigquery_project, args.bigquery_dataset, "salesforce_order_daily_view", "date_day", ) db_order_dates = fetch_databend_values( args.databend_input_db, "salesforce_order_daily_view", "date_day", ) common_billing_months = common_values(bq_billing_months, db_billing_months) common_order_dates = common_values(bq_order_dates, db_order_dates) billing_months = resolve_selector(args.billing_month, common_billing_months, "billing_month") order_dates = resolve_selector(args.order_date, common_order_dates, "order_date") cases = [ CaseSpec(billing_month=billing_month, order_date=order_date) for billing_month in billing_months for order_date in order_dates ] return cases, { "common_billing_months": common_billing_months, "common_order_dates": common_order_dates, "bigquery_only_billing_months": value_gaps(bq_billing_months, db_billing_months), "databend_only_billing_months": value_gaps(db_billing_months, bq_billing_months), "bigquery_only_order_dates": value_gaps(bq_order_dates, db_order_dates), "databend_only_order_dates": value_gaps(db_order_dates, bq_order_dates), "selected_billing_months": billing_months, "selected_order_dates": order_dates, } def fetch_bigquery_inputs( project: str, dataset: str, billing_months: list[str], order_dates: list[str], ) -> tuple[pd.DataFrame, pd.DataFrame]: billing_query = f""" SELECT * FROM `{project}.{dataset}.salesforce_org_monthly_revrec` WHERE billing_month IN ({quote_sql_values(billing_months)}) """ billing = pd.DataFrame( fetch_bigquery_query( project, f""" SELECT * FROM `{project}.{dataset}.salesforce_org_monthly_revrec` WHERE billing_month IN ({quote_sql_values(billing_months)}) ORDER BY billing_month, _row """, count_query=billing_query, label="salesforce_org_monthly_revrec parity inputs", ) ) orders_query = f""" SELECT * FROM `{project}.{dataset}.salesforce_order_daily_view` WHERE date_day IN ({quote_sql_values(order_dates)}) """ orders = pd.DataFrame( fetch_bigquery_query( project, f""" SELECT * FROM `{project}.{dataset}.salesforce_order_daily_view` WHERE date_day IN ({quote_sql_values(order_dates)}) ORDER BY date_day, order_id """, count_query=orders_query, label="salesforce_order_daily_view parity inputs", ) ) return coerce_inputs(orders, billing) def fetch_databend_inputs( input_db: str, billing_months: list[str], order_dates: list[str], ) -> tuple[pd.DataFrame, pd.DataFrame]: cur = _databend_cursor() billing = databend_io.read_df( cur, f""" SELECT * FROM {input_db}.salesforce_org_monthly_revrec WHERE billing_month IN ({quote_sql_values(billing_months)}) ORDER BY billing_month, _row """, numeric=job._BILLING_NUMERIC, date=job._BILLING_DATE, ) orders = databend_io.read_df( cur, f""" SELECT * FROM {input_db}.salesforce_order_daily_view WHERE date_day IN ({quote_sql_values(order_dates)}) ORDER BY date_day, order_id """, boolean=job._ORDER_BOOLEAN, ) return orders, billing def coerce_inputs(df_orders: pd.DataFrame, df_billing: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: for frame in (df_orders, df_billing): for column in DATE_COLUMNS & set(frame.columns): frame[column] = pd.to_datetime(frame[column], errors="coerce").dt.date for column in BOOL_COLUMNS & set(frame.columns): frame[column] = frame[column].map(parse_bool) for column in job._BILLING_NUMERIC & set(df_billing.columns): df_billing[column] = pd.to_numeric(df_billing[column], errors="coerce") for column in ORDER_NUMERIC & set(df_orders.columns): df_orders[column] = pd.to_numeric(df_orders[column], errors="coerce") return df_orders, df_billing def parse_bool(value: Any) -> bool | None: if value is None: return None try: if pd.isna(value): return None except (TypeError, ValueError): pass if isinstance(value, bool): return value text = str(value).strip().lower() if text in {"true", "t", "1", "yes"}: return True if text in {"false", "f", "0", "no"}: return False return bool(value) def filter_inputs( df_orders: pd.DataFrame, df_billing: pd.DataFrame, *, billing_month: str, order_date: str, revenue: str, ) -> tuple[pd.DataFrame, pd.DataFrame]: parsed_order_date = pd.to_datetime(order_date).date() df_orders = df_orders[ (df_orders["date_day"] == parsed_order_date) & (df_orders["order_service_start_date"] <= parsed_order_date) ].copy() df_billing = df_billing[df_billing["billing_month"].astype(str) == billing_month].copy() if revenue == "recognised": df_billing = df_billing[ df_billing["actual_revenue_recognition_date"].notna() & (df_billing["actual_revenue_recognition_date"] != "") ].copy() else: df_billing = df_billing[ df_billing["actual_revenue_recognition_date"].isna() | (df_billing["actual_revenue_recognition_date"] == "") ].copy() return df_orders, df_billing def run_expected_from_inputs( df_orders: pd.DataFrame, df_billing: pd.DataFrame, *, billing_month: str, order_date: str, revenue: str, ) -> tuple[pd.DataFrame, pd.DataFrame, dict[str, int]]: filtered_orders, filtered_billing = filter_inputs( df_orders, df_billing, billing_month=billing_month, order_date=order_date, revenue=revenue, ) collections, orders = job.process(df_orders=filtered_orders, df_billing=filtered_billing) return collections, orders, { "billing": len(filtered_billing), "orders": len(filtered_orders), } def normalize(column: str, value: Any) -> Any: if value is None or value is pd.NaT: return None if isinstance(value, float) and math.isnan(value): return None try: if pd.isna(value): return None except (TypeError, ValueError): pass if column in DATE_COLUMNS or column == "actual_revenue_recognition_date": return str(value)[:10] if column in BOOL_COLUMNS: return parse_bool(value) if column in OUTPUT_NUMERIC: try: return Decimal(str(value)) except InvalidOperation: return value return str(value) def key_rows(df: pd.DataFrame, key_columns: tuple[str, ...], label: str) -> dict[tuple[str, ...], dict[str, Any]]: result: dict[tuple[str, ...], dict[str, Any]] = {} duplicates: list[tuple[str, ...]] = [] for row in df.to_dict("records"): key = tuple(str(normalize(column, row.get(column))) for column in key_columns) if key in result: duplicates.append(key) result[key] = row if duplicates: raise RuntimeError(f"{label} has duplicate key {key_columns}: {duplicates[0]}") return result def values_match(column: str, left: Any, right: Any, tolerance: Decimal) -> bool: left = normalize(column, left) right = normalize(column, right) if left is None or right is None: return left is None and right is None if column in OUTPUT_NUMERIC and isinstance(left, Decimal) and isinstance(right, Decimal): return abs(left - right) <= tolerance return left == right def compare_frames( bq: pd.DataFrame, db: pd.DataFrame, *, key_columns: tuple[str, ...], tolerance: Decimal, sample_limit: int, ) -> dict[str, Any]: bq_by_key = key_rows(bq, key_columns, "bigquery") db_by_key = key_rows(db, key_columns, "databend") bq_keys = set(bq_by_key) db_keys = set(db_by_key) common = sorted(bq_keys & db_keys) columns = [column for column in bq.columns if column not in key_columns] summaries = {column: FieldSummary() for column in columns} for key in common: for column, summary in summaries.items(): if values_match(column, bq_by_key[key].get(column), db_by_key[key].get(column), tolerance): summary.matches += 1 else: summary.mismatches += 1 summary.add_example( key, normalize(column, bq_by_key[key].get(column)), normalize(column, db_by_key[key].get(column)), sample_limit, ) return { "row_counts": {"bigquery": len(bq), "databend": len(db), "common": len(common)}, "missing_keys": { "bigquery_only": sample_keys(sorted(bq_keys - db_keys), sample_limit), "databend_only": sample_keys(sorted(db_keys - bq_keys), sample_limit), }, "fields": { column: { "matches": summary.matches, "mismatches": summary.mismatches, "examples": jsonable(summary.examples or []), } for column, summary in summaries.items() }, } def sample_keys(keys: list[tuple[str, ...]], sample_limit: int) -> dict[str, Any]: return {"count": len(keys), "examples": [list(key) for key in keys[:sample_limit]]} def jsonable(value: Any) -> Any: if isinstance(value, Decimal): return float(value) if value.is_finite() else str(value) if isinstance(value, dict): return {key: jsonable(item) for key, item in value.items()} if isinstance(value, list): return [jsonable(item) for item in value] if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): return str(value) return value def has_failures(summary: dict[str, Any]) -> bool: return case_has_failures(summary, allow_empty_matches=False) def case_has_failures(summary: dict[str, Any], *, allow_empty_matches: bool) -> bool: if allow_empty_matches and case_is_allowed_empty_match(summary): return False for output in ("collections", "orders"): item = summary[output] if item["row_counts"]["bigquery"] == 0 or item["row_counts"]["databend"] == 0: return True if item["missing_keys"]["bigquery_only"]["count"] or item["missing_keys"]["databend_only"]["count"]: return True if any(field["mismatches"] for field in item["fields"].values()): return True return False def case_is_empty_match(summary: dict[str, Any]) -> bool: for output in ("collections", "orders"): item = summary[output] if item["row_counts"] != {"bigquery": 0, "databend": 0, "common": 0}: return False if item["missing_keys"]["bigquery_only"]["count"] or item["missing_keys"]["databend_only"]["count"]: return False if any(field["mismatches"] for field in item["fields"].values()): return False return True def case_is_allowed_empty_match(summary: dict[str, Any]) -> bool: input_rows = summary.get("input_rows", {}) return ( summary.get("revenue") == "recognised" and case_is_empty_match(summary) and input_rows.get("bigquery", {}).get("billing") == 0 and input_rows.get("databend", {}).get("billing") == 0 ) def output_missing_key_count(output_summary: dict[str, Any]) -> int: return ( output_summary["missing_keys"]["bigquery_only"]["count"] + output_summary["missing_keys"]["databend_only"]["count"] ) def output_field_mismatch_count(output_summary: dict[str, Any]) -> int: return sum(field["mismatches"] for field in output_summary["fields"].values()) def case_summary(case: dict[str, Any], *, allow_empty_matches: bool = False) -> dict[str, Any]: empty_match = case_is_empty_match(case) return { "billing_month": case["billing_month"], "order_date": case["order_date"], "revenue": case["revenue"], "input_rows": case.get("input_rows", {}), "collections_rows": case["collections"]["row_counts"], "collections_missing_keys": output_missing_key_count(case["collections"]), "collections_field_mismatches": output_field_mismatch_count(case["collections"]), "orders_rows": case["orders"]["row_counts"], "orders_missing_keys": output_missing_key_count(case["orders"]), "orders_field_mismatches": output_field_mismatch_count(case["orders"]), "empty_match": empty_match, "failed": case_has_failures(case, allow_empty_matches=allow_empty_matches), } def format_summary_table(summaries: list[dict[str, Any]]) -> str: headers = [ "billing_month", "order_date", "revenue", "collections", "collections_missing_keys", "collections_field_mismatches", "orders", "orders_missing_keys", "orders_field_mismatches", "status", ] rows: list[list[str]] = [] for summary in summaries: collections_rows = summary["collections_rows"] orders_rows = summary["orders_rows"] rows.append( [ summary["billing_month"], summary["order_date"], summary["revenue"], f"{collections_rows['bigquery']}/{collections_rows['databend']}", str(summary["collections_missing_keys"]), str(summary["collections_field_mismatches"]), f"{orders_rows['bigquery']}/{orders_rows['databend']}", str(summary["orders_missing_keys"]), str(summary["orders_field_mismatches"]), "FAIL" if summary["failed"] else "PASS_EMPTY" if summary["empty_match"] else "PASS", ] ) widths = [len(header) for header in headers] for row in rows: widths = [max(width, len(value)) for width, value in zip(widths, row, strict=True)] line = " | ".join(header.ljust(width) for header, width in zip(headers, widths, strict=True)) sep = "-+-".join("-" * width for width in widths) body = [" | ".join(value.ljust(width) for value, width in zip(row, widths, strict=True)) for row in rows] return "\n".join([line, sep, *body]) def build_summary( discovery: dict[str, Any], cases: list[dict[str, Any]], *, allow_empty_matches: bool, summary_only: bool, ) -> dict[str, Any]: case_summaries = [case_summary(case, allow_empty_matches=allow_empty_matches) for case in cases] summary = {"discovery": discovery, "case_summaries": case_summaries} if summary_only: failing_cases = [ case for case in cases if case_has_failures(case, allow_empty_matches=allow_empty_matches) ] if failing_cases: summary["failing_cases"] = failing_cases else: summary["cases"] = cases return summary def run_case( args: argparse.Namespace, case: CaseSpec, revenue: str, *, bq_orders: pd.DataFrame, bq_billing: pd.DataFrame, db_orders: pd.DataFrame, db_billing: pd.DataFrame, ) -> dict[str, Any]: bq_collections, bq_orders_output, bq_input_rows = run_expected_from_inputs( bq_orders, bq_billing, billing_month=case.billing_month, order_date=case.order_date, revenue=revenue, ) db_collections, db_orders_output, db_input_rows = run_expected_from_inputs( db_orders, db_billing, billing_month=case.billing_month, order_date=case.order_date, revenue=revenue, ) tolerance = Decimal(args.numeric_tolerance) return { "billing_month": case.billing_month, "order_date": case.order_date, "revenue": revenue, "numeric_tolerance": str(tolerance), "input_rows": { "bigquery": bq_input_rows, "databend": db_input_rows, }, "collections": compare_frames( bq_collections, db_collections, key_columns=COLLECTION_KEY, tolerance=tolerance, sample_limit=args.sample_limit, ), "orders": compare_frames( bq_orders_output, db_orders_output, key_columns=ORDER_KEY, tolerance=tolerance, sample_limit=args.sample_limit, ), } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--billing-month", default="latest", help="YYYY-MM, latest, or all") parser.add_argument("--order-date", default="latest", help="YYYY-MM-DD, latest, or all") parser.add_argument("--revenue", choices=["recognised", "predicted", "all"], default="all") parser.add_argument("--bigquery-project", default=DEFAULT_BIGQUERY_PROJECT) parser.add_argument("--bigquery-dataset", default=DEFAULT_BIGQUERY_DATASET) parser.add_argument("--databend-input-db", default=DEFAULT_DATABEND_INPUT_DB) parser.add_argument("--numeric-tolerance", default="0.01") parser.add_argument("--sample-limit", type=int, default=5) parser.add_argument( "--list-cases", action="store_true", help="Discover selected comparable cases without running parity", ) parser.add_argument( "--allow-cartesian", action="store_true", help="Allow more than one billing month and more than one order date in one run", ) parser.add_argument( "--no-summary-table", action="store_true", help="Suppress the compact stderr summary table", ) parser.add_argument( "--summary-only", action="store_true", help="Emit discovery and compact summaries, plus full details for any failing cases", ) parser.add_argument( "--allow-empty-matches", action="store_true", help=( "Treat recognised cases with zero filtered billing inputs and empty outputs in both " "warehouses as passing matrix cases" ), ) args = parser.parse_args() logging.getLogger(job.__name__).setLevel(logging.ERROR) revenues = ["recognised", "predicted"] if args.revenue == "all" else [args.revenue] selected_cases, discovery = discover_common_cases(args) if args.list_cases: print( json.dumps( jsonable({"discovery": discovery, "cases": [case.__dict__ for case in selected_cases]}), indent=2, sort_keys=True, ) ) return if ( len(discovery["selected_billing_months"]) > 1 and len(discovery["selected_order_dates"]) > 1 and not args.allow_cartesian ): raise SystemExit( "Refusing to run a multi-month x multi-date Cartesian parity matrix without " "--allow-cartesian. Use --order-date latest for the normal broad proof." ) bq_orders, bq_billing = fetch_bigquery_inputs( args.bigquery_project, args.bigquery_dataset, discovery["selected_billing_months"], discovery["selected_order_dates"], ) db_orders, db_billing = fetch_databend_inputs( args.databend_input_db, discovery["selected_billing_months"], discovery["selected_order_dates"], ) cases = [ run_case( args, case, revenue, bq_orders=bq_orders, bq_billing=bq_billing, db_orders=db_orders, db_billing=db_billing, ) for case in selected_cases for revenue in revenues ] summary = build_summary( discovery, cases, allow_empty_matches=args.allow_empty_matches, summary_only=args.summary_only, ) case_summaries = summary["case_summaries"] if not args.no_summary_table: print(format_summary_table(case_summaries), file=sys.stderr) print(json.dumps(jsonable(summary), indent=2, sort_keys=True)) if any(case_has_failures(case, allow_empty_matches=args.allow_empty_matches) for case in cases): raise SystemExit(1) if __name__ == "__main__": main()