add cost allocation for zerobus ingest, mv refresh and predictive op

full-path-realtime/quotes/databricks/README.md

@@ -190,6 +190,8 @@ substitute for the producer's file hashes and canonical task manifest.
   catalog and schema are administrator prerequisites.
 - `queries_mv.sql`: the four canonical dashboard statements in fixed order.
 - `queries_raw.sql`: the two canonical drill-down statements in fixed order.
+- `export_allocation_details.sql`: detailed non-aggregated MV refresh,
+  Predictive Optimization, and Zerobus allocation exports.
 - `dbx_common.py`: standard-library REST, SQL rendering/splitting, bounded
   result retrieval, atomic evidence writes, configuration, and redaction.
 - `apply_ddl.py`: standard-library destructive-confirmation wrapper that

full-path-realtime/quotes/databricks/RESULT_COLLECTION_REFERENCE.md

@@ -206,6 +206,27 @@ After each observation, the monitor atomically replaces this file with the
 latest compact sample. It is a current-state snapshot, not an aggregate over
 the JSONL history.
 
+## `freshness/mv_refresh_allocation.csv`
+
+**How data was collected**
+
+- Exact SQL: query 1 in `export_allocation_details.sql`.
+- Direct sources: `system.billing.usage` and
+  `system.billing.list_prices`.
+- Attribution: UC catalog/schema/table metadata is used to discover every
+  `dlt_pipeline_id` for the target MV; all rows for those pipeline IDs are then
+  included, including rows whose UC table fields are null.
+- Allocation: billing intervals are clipped to
+  `[run_start, producer_finished)` and `usage_quantity` is prorated by exact
+  microsecond overlap using `DECIMAL(38, 18)`.
+- Cost: each detailed allocated DBU line is joined to the one effective list
+  price covering its complete source billing interval.
+
+The CSV contains one billing allocation line per provider usage record. It is
+not redundant with freshness JSONL: freshness records MV data lag and refresh
+watermarks, while this file records refresh billing DBUs and list cost over
+time. Public packaging pseudonymizes billing record and pipeline IDs.
+
 ## `ingest/ingest_metrics.jsonl`
 
 **How data was collected**
@@ -294,6 +315,41 @@ close and inspection succeeded, unacknowledged batch/row counts, persisted
 diagnostic artifacts, and inspection errors. The file exists even on a worker
 failure so that ambiguous durability is visible.
 
+## `ingest/zerobus_ingest_allocation.csv`
+
+**How data was collected**
+
+- Exact SQL: query 3 in `export_allocation_details.sql`.
+- Direct source: `system.billing.usage`.
+- Attribution:
+  `product_features.lakeflow_connect.zerobus_request_type = 'GRPC'` plus the
+  raw table's Unity Catalog table ID.
+- Allocation: DBU intervals are clipped to
+  `[run_start, producer_finished)` and prorated by exact microsecond overlap
+  using `DECIMAL(38, 18)`.
+
+The CSV contains one signed billing line per interval, with source and
+benchmark-allocated DBUs. It is not redundant with ingest metrics or provider
+reconciliation: those contain row/byte throughput and durability, not billed
+DBUs. Public packaging pseudonymizes billing record and table IDs.
+
+## `ingest/predictive_optimization_allocation.csv`
+
+**How data was collected**
+
+- Exact SQL: query 2 in `export_allocation_details.sql`.
+- Direct source:
+  `system.storage.predictive_optimization_operations_history`.
+- Filter: target catalog/schema/raw/MV names and operation intervals
+  overlapping `[run_start, producer_finished)`.
+- Values: operation type/status, start/end, provider operation metrics, and
+  `usage_quantity` reported in `ESTIMATED_DBU`.
+
+This CSV is the sole packaged operation-level Predictive Optimization detail.
+The compact clustering JSON retains table configuration and operation
+summaries/counts but no longer duplicates the operation array. Public
+packaging pseudonymizes metastore and operation IDs.
+
 ## `evidence/provider_reconciliation.json`
 
 **How data was collected**
@@ -352,8 +408,10 @@ file reports type counts, class counts, and the derived full-refresh count.
 
 The file keeps table format/location metadata, clustering columns, relevant
 Delta features and properties, compression settings, configured/effective
-Predictive Optimization state, operation summaries, and the count of
-clustering/optimize operations. It does not inspect table contents.
+Predictive Optimization state, aggregate operation summaries, and the count
+of clustering/optimize operations. Detailed operation rows live only in
+`ingest/predictive_optimization_allocation.csv`. It does not inspect table
+contents.
 
 ## `validation/preflight.json`
 

full-path-realtime/quotes/databricks/RESULT_FIELD_GLOSSARY.md

@@ -171,6 +171,27 @@ The progress file is the latest snapshot; each JSONL line has the same shape.
 - `statement_ids.zerobus_ingest`, `statement_ids.raw_history`, `statement_ids.mv_events`: provider evidence statements.
 - `errors[]`: shared `error_entry` objects.
 
+## `freshness/mv_refresh_allocation.csv`
+
+- `usage_start_time`, `usage_end_time`: complete provider billing interval.
+- `allocated_start`, `allocated_end`: interval clipped to the benchmark's
+  active-ingestion window.
+- `record_id`: pseudonymized billing-record identifier.
+- `record_type`: `ORIGINAL`, `RETRACTION`, or `RESTATEMENT`; quantities retain
+  their provider-exported sign.
+- `pipeline_id`: pseudonymized MV pipeline identifier.
+- `catalog_name`, `schema_name`, `table_name`: optional UC table attribution.
+  They can be null even when the pipeline ID is present.
+- `billing_origin_product`, `sku_name`: provider billing classification and
+  SKU.
+- `source_dbu`: DBUs for the complete source billing interval.
+- `allocated_dbu`: exact prorated DBUs within the benchmark window.
+- `currency_code`, `price_per_dbu`: effective list-price currency and unit
+  price.
+- `cost`: `allocated_dbu * price_per_dbu`.
+- `custom_tags`: provider custom-tag map.
+- `ingestion_date`: date the billing row entered the billing system.
+
 ## `ingest/ingest_metrics.jsonl`
 
 Each line is one compact producer metrics snapshot.
@@ -265,6 +286,34 @@ This pattern contains every field defined for `ingest/ingest_progress.json` plus
 - `artifacts[]`: preserved unacknowledged Arrow batches using the artifact fields under `stream_status`.
 - `inspection_error`: optional redacted inspection failure.
 
+## `ingest/zerobus_ingest_allocation.csv`
+
+- `usage_start_time`, `usage_end_time`: complete provider billing interval.
+- `allocated_start`, `allocated_end`: interval clipped to active ingestion.
+- `record_id`: pseudonymized billing-record identifier.
+- `record_type`: signed billing record type.
+- `table_id`: pseudonymized Unity Catalog target table ID.
+- `zerobus_request_type`: protocol/request attribution, such as `GRPC`.
+- `billing_origin_product`, `sku_name`: provider billing product and SKU.
+- `source_dbu`: DBUs for the complete billing interval.
+- `allocated_dbu`: exact prorated DBUs within the benchmark window.
+- `custom_tags`: provider custom-tag map.
+- `ingestion_date`: billing ingestion date.
+
+## `ingest/predictive_optimization_allocation.csv`
+
+- `metastore_name`: pseudonymized metastore identifier.
+- `catalog_name`, `schema_name`, `table_name`: operation target.
+- `operation_id`: pseudonymized operation identifier.
+- `operation_type`: operation class, such as `CLUSTERING` or `ANALYZE`.
+- `operation_status`: provider operation outcome.
+- `start_time`, `end_time`: operation interval.
+- `usage_unit`: provider unit, `ESTIMATED_DBU` for this source.
+- `estimated_dbu`: provider-estimated operation DBUs; not authoritative billed
+  DBUs.
+- `operation_metrics`: provider JSON metrics such as files/bytes scanned,
+  removed, or clustered.
+
 ## `evidence/provider_reconciliation.json`
 
 - `schema_version`, `run_id`, `collected_at`: schema revision, correlated run identifier, and collection time.
@@ -300,10 +349,9 @@ This pattern contains every field defined for `ingest/ingest_progress.json` plus
 - `predictive_optimization.summary.operation_count`, `predictive_optimization.summary.operation_counts_by_type.{type}`: operation totals.
 - `predictive_optimization.summary.usage_by_unit.{unit}`: usage quantity by provider unit.
 - `predictive_optimization.clustering_operation_count`: clustering-class operation count.
-- `predictive_optimization.operations[]`: compact provider operations.
-- `operations[].table_name`, `operations[].operation_id`, `operations[].operation_type`, `operations[].operation_status`: operation identity and status.
-- `operations[].start_time`, `operations[].end_time`, `operations[].usage_unit`, `operations[].usage_quantity`: operation bounds and usage.
-- `operations[].operation_metrics.{metric}`: decoded provider metric map.
+
+Detailed operation rows are stored only in
+`ingest/predictive_optimization_allocation.csv`.
 
 ## `validation/preflight.json`
 

full-path-realtime/quotes/databricks/export_allocation_details.sql

@@ -0,0 +1,202 @@
+-- Detailed, non-aggregated exports used by the compact result package.
+-- Replace every <...> placeholder before running on a SQL warehouse.
+-- Public result packaging pseudonymizes provider record/resource IDs.
+
+-- ---------------------------------------------------------------------------
+-- 1. MV refresh billing detail
+-- Output: freshness/mv_refresh_allocation.csv
+-- ---------------------------------------------------------------------------
+
+WITH
+params AS (
+    SELECT
+        CAST('<run-start-utc>' AS TIMESTAMP) AS run_start,
+        CAST('<producer-finished-utc>' AS TIMESTAMP) AS run_end
+),
+
+mv_pipeline_ids AS (
+    SELECT DISTINCT usage_metadata.dlt_pipeline_id AS pipeline_id
+    FROM system.billing.usage
+    CROSS JOIN params
+    WHERE usage_metadata.uc_table_catalog = '<catalog>'
+      AND usage_metadata.uc_table_schema = '<schema>'
+      AND usage_metadata.uc_table_name =
+          '<catalog>.<schema>.<materialized-view>'
+      AND usage_metadata.dlt_pipeline_id IS NOT NULL
+      AND usage_start_time < run_end
+      AND usage_end_time > run_start
+),
+
+usage_lines AS (
+    SELECT
+        u.*,
+        greatest(u.usage_start_time, p.run_start) AS allocated_start,
+        least(u.usage_end_time, p.run_end) AS allocated_end
+    FROM system.billing.usage AS u
+    CROSS JOIN params AS p
+    WHERE u.usage_metadata.dlt_pipeline_id IN (
+        SELECT pipeline_id FROM mv_pipeline_ids
+    )
+      AND u.usage_unit = 'DBU'
+      AND u.usage_start_time < p.run_end
+      AND u.usage_end_time > p.run_start
+),
+
+allocated AS (
+    SELECT
+        *,
+        usage_quantity
+        * CAST(
+            timestampdiff(
+                MICROSECOND,
+                allocated_start,
+                allocated_end
+            ) AS DECIMAL(38, 18)
+        )
+        / nullif(
+            CAST(
+                timestampdiff(
+                    MICROSECOND,
+                    usage_start_time,
+                    usage_end_time
+                ) AS DECIMAL(38, 18)
+            ),
+            0
+        ) AS allocated_dbu
+    FROM usage_lines
+    WHERE allocated_end > allocated_start
+)
+
+SELECT
+    u.usage_start_time,
+    u.usage_end_time,
+    u.allocated_start,
+    u.allocated_end,
+    u.record_id,
+    u.record_type,
+    u.usage_metadata.dlt_pipeline_id AS pipeline_id,
+    u.usage_metadata.uc_table_catalog AS catalog_name,
+    u.usage_metadata.uc_table_schema AS schema_name,
+    u.usage_metadata.uc_table_name AS table_name,
+    u.billing_origin_product,
+    u.sku_name,
+    u.usage_quantity AS source_dbu,
+    u.allocated_dbu,
+    p.currency_code,
+    coalesce(
+        p.pricing.effective_list.default,
+        p.pricing.default
+    ) AS price_per_dbu,
+    u.allocated_dbu * coalesce(
+        p.pricing.effective_list.default,
+        p.pricing.default
+    ) AS cost,
+    u.custom_tags,
+    u.ingestion_date
+FROM allocated AS u
+JOIN system.billing.list_prices AS p
+  ON lower(p.sku_name) = lower(u.sku_name)
+ AND lower(p.cloud) = lower(u.cloud)
+ AND lower(p.usage_unit) = lower(u.usage_unit)
+ AND p.price_start_time <= u.usage_start_time
+ AND (
+      p.price_end_time IS NULL
+      OR u.usage_end_time <= p.price_end_time
+ )
+ORDER BY u.usage_start_time, u.record_id;
+
+
+-- ---------------------------------------------------------------------------
+-- 2. Predictive Optimization operation detail
+-- Output: ingest/predictive_optimization_allocation.csv
+-- `estimated_dbu` is provider-estimated, not authoritative billed DBU.
+-- ---------------------------------------------------------------------------
+
+SELECT
+    metastore_name,
+    catalog_name,
+    schema_name,
+    table_name,
+    operation_id,
+    operation_type,
+    operation_status,
+    start_time,
+    end_time,
+    usage_unit,
+    usage_quantity AS estimated_dbu,
+    operation_metrics
+FROM system.storage.predictive_optimization_operations_history
+WHERE catalog_name = '<catalog>'
+  AND schema_name = '<schema>'
+  AND table_name IN ('<raw-table>', '<materialized-view>')
+  AND start_time < CAST('<producer-finished-utc>' AS TIMESTAMP)
+  AND coalesce(
+        end_time,
+        CAST('<producer-finished-utc>' AS TIMESTAMP)
+      ) > CAST('<run-start-utc>' AS TIMESTAMP)
+ORDER BY start_time, operation_id;
+
+
+-- ---------------------------------------------------------------------------
+-- 3. Zerobus DBU detail
+-- Output: ingest/zerobus_ingest_allocation.csv
+-- Find the UC table ID in DESCRIBE DETAIL properties['io.unitycatalog.tableId'].
+-- ---------------------------------------------------------------------------
+
+WITH
+params AS (
+    SELECT
+        CAST('<run-start-utc>' AS TIMESTAMP) AS run_start,
+        CAST('<producer-finished-utc>' AS TIMESTAMP) AS run_end
+),
+
+usage_lines AS (
+    SELECT
+        u.*,
+        greatest(u.usage_start_time, p.run_start) AS allocated_start,
+        least(u.usage_end_time, p.run_end) AS allocated_end
+    FROM system.billing.usage AS u
+    CROSS JOIN params AS p
+    WHERE u.product_features.lakeflow_connect.zerobus_request_type = 'GRPC'
+      AND u.usage_metadata.table_id = '<unity-catalog-table-id>'
+      AND u.usage_unit = 'DBU'
+      AND u.usage_start_time < p.run_end
+      AND u.usage_end_time > p.run_start
+)
+
+SELECT
+    usage_start_time,
+    usage_end_time,
+    allocated_start,
+    allocated_end,
+    record_id,
+    record_type,
+    usage_metadata.table_id AS table_id,
+    product_features.lakeflow_connect.zerobus_request_type
+        AS zerobus_request_type,
+    billing_origin_product,
+    sku_name,
+    usage_quantity AS source_dbu,
+    usage_quantity
+    * CAST(
+        timestampdiff(
+            MICROSECOND,
+            allocated_start,
+            allocated_end
+        ) AS DECIMAL(38, 18)
+    )
+    / nullif(
+        CAST(
+            timestampdiff(
+                MICROSECOND,
+                usage_start_time,
+                usage_end_time
+            ) AS DECIMAL(38, 18)
+        ),
+        0
+    ) AS allocated_dbu,
+    custom_tags,
+    ingestion_date
+FROM usage_lines
+WHERE allocated_end > allocated_start
+ORDER BY usage_start_time, record_id;