Dashboard As Code for Reconcile

src/databricks/labs/remorph/resources/reconcile/dashboards/0_1_recon_main.sql

@@ -0,0 +1,31 @@
+select
+main.recon_id,
+main.source_type,
+main.report_type,
+main.source_table.`catalog` as source_catalog,
+main.source_table.`schema` as source_schema,
+main.source_table.table_name as source_table_name,
+CASE
+        WHEN COALESCE(MAIN.SOURCE_TABLE.CATALOG, '') <> '' THEN CONCAT(MAIN.SOURCE_TABLE.CATALOG, '.', MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+        ELSE CONCAT(MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+    END AS source_table,
+main.target_table.`catalog` as target_catalog,
+main.target_table.`schema` as target_schema,
+main.target_table.table_name as target_table_name,
+CONCAT(MAIN.TARGET_TABLE.CATALOG, '.', MAIN.TARGET_TABLE.SCHEMA, '.', MAIN.TARGET_TABLE.TABLE_NAME) AS target_table,
+metrics.run_metrics.status as status,
+metrics.run_metrics.exception_message as exception,
+metrics.recon_metrics.row_comparison.missing_in_source as missing_in_source,
+metrics.recon_metrics.row_comparison.missing_in_target as missing_in_target,
+metrics.recon_metrics.column_comparison.absolute_mismatch as absolute_mismatch,
+metrics.recon_metrics.column_comparison.threshold_mismatch as threshold_mismatch,
+metrics.recon_metrics.column_comparison.mismatch_columns as mismatch_columns,
+metrics.recon_metrics.schema_comparison as schema_comparison,
+metrics.run_metrics.run_by_user as executed_by,
+main.start_ts as start_ts,
+main.end_ts as end_ts
+from IDENTIFIER(:catalog || '.' || :schema || '.main' ) main
+inner join
+IDENTIFIER(:catalog || '.' || :schema || '.metrics' ) metrics
+on main.recon_table_id = metrics.recon_table_id
+order by metrics.inserted_ts desc, main.recon_id, main.target_table.table_name

src/databricks/labs/remorph/resources/reconcile/dashboards/1_0_recon_details_schema.sql

@@ -0,0 +1,31 @@
+with tmp as (select
+recon_table_id,
+inserted_ts,
+explode(data) as schema_data
+from IDENTIFIER(:catalog || '.' || :schema || '.details' )
+where recon_type='schema'
+)
+select
+main.recon_id,
+main.source_table.`catalog` as source_catalog,
+main.source_table.`schema` as source_schema,
+main.source_table.table_name as source_table_name,
+CASE
+        WHEN COALESCE(MAIN.SOURCE_TABLE.CATALOG, '') <> '' THEN CONCAT(MAIN.SOURCE_TABLE.CATALOG, '.', MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+        ELSE CONCAT(MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+    END AS source_table,
+main.target_table.`catalog` as target_catalog,
+main.target_table.`schema` as target_schema,
+main.target_table.table_name as target_table_name,
+CONCAT(MAIN.TARGET_TABLE.CATALOG, '.', MAIN.TARGET_TABLE.SCHEMA, '.', MAIN.TARGET_TABLE.TABLE_NAME) AS target_table,
+schema_data['source_column'] as source_column,
+schema_data['source_datatype'] as source_datatype,
+schema_data['databricks_column'] as databricks_column,
+schema_data['databricks_datatype'] as databricks_datatype,
+schema_data['is_valid'] as is_valid
+from
+IDENTIFIER(:catalog || '.' || :schema || '.main' ) main
+inner join
+tmp
+on main.recon_table_id = tmp.recon_table_id
+order by tmp.inserted_ts desc, main.recon_id, main.target_table

src/databricks/labs/remorph/resources/reconcile/dashboards/1_1_recon_details_pivot.sql

@@ -0,0 +1,22 @@
+with tmp as (select recon_table_id, inserted_ts ,recon_type, explode(data) as data,
+row_number() over(partition by recon_table_id,recon_type order by recon_table_id) as rn
+from IDENTIFIER(:catalog || '.' || :schema || '.details' )
+where recon_type != 'schema')
+select main.recon_id,
+main.source_table.`catalog` as source_catalog,
+main.source_table.`schema` as source_schema,
+main.source_table.table_name as source_table_name,
+CASE
+        WHEN COALESCE(MAIN.SOURCE_TABLE.CATALOG, '') <> '' THEN CONCAT(MAIN.SOURCE_TABLE.CATALOG, '.', MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+        ELSE CONCAT(MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+    END AS source_table,
+main.target_table.`catalog` as target_catalog,
+main.target_table.`schema` as target_schema,
+main.target_table.table_name as target_table_name,
+CONCAT(MAIN.TARGET_TABLE.CATALOG, '.', MAIN.TARGET_TABLE.SCHEMA, '.', MAIN.TARGET_TABLE.TABLE_NAME) AS target_table,
+recon_type, key, value, rn
+from tmp
+inner join
+IDENTIFIER(:catalog || '.' || :schema || '.main' ) main
+on main.recon_table_id = tmp.recon_table_id
+lateral view explode(data) exploded_data AS key, value

src/databricks/labs/remorph/resources/reconcile/dashboards/2_1_failed_table_list.sql

@@ -0,0 +1,26 @@
+select
+main.recon_id,
+main.source_type,
+main.report_type,
+main.source_table.`catalog` as source_catalog,
+main.source_table.`schema` as source_schema,
+main.source_table.table_name as source_table_name,
+CASE
+        WHEN COALESCE(MAIN.SOURCE_TABLE.CATALOG, '') <> '' THEN CONCAT(MAIN.SOURCE_TABLE.CATALOG, '.', MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+        ELSE CONCAT(MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+    END AS source_table,
+main.target_table.`catalog` as target_catalog,
+main.target_table.`schema` as target_schema,
+main.target_table.table_name as target_table_name,
+CONCAT(MAIN.TARGET_TABLE.CATALOG, '.', MAIN.TARGET_TABLE.SCHEMA, '.', MAIN.TARGET_TABLE.TABLE_NAME) AS target_table,
+metrics.run_metrics.status as status,
+metrics.run_metrics.run_by_user as executed_by,
+main.start_ts as start_ts,
+main.end_ts as end_ts,
+date(main.start_ts) as start_date
+from IDENTIFIER(:catalog || '.' || :schema || '.main' ) main
+inner join
+IDENTIFIER(:catalog || '.' || :schema || '.metrics' ) metrics
+on main.recon_table_id = metrics.recon_table_id
+where metrics.run_metrics.status = false
+order by metrics.inserted_ts desc, main.recon_id, main.target_table.table_name

src/databricks/labs/remorph/resources/reconcile/dashboards/2_2_success_table_list.sql

@@ -0,0 +1,20 @@
+select
+main.recon_id,
+main.source_type,
+main.report_type,
+CASE
+        WHEN COALESCE(MAIN.SOURCE_TABLE.CATALOG, '') <> '' THEN CONCAT(MAIN.SOURCE_TABLE.CATALOG, '.', MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+        ELSE CONCAT(MAIN.SOURCE_TABLE.SCHEMA, '.', MAIN.SOURCE_TABLE.TABLE_NAME)
+    END AS source_table,
+CONCAT(MAIN.TARGET_TABLE.CATALOG, '.', MAIN.TARGET_TABLE.SCHEMA, '.', MAIN.TARGET_TABLE.TABLE_NAME) AS target_table,
+metrics.run_metrics.status as status,
+metrics.run_metrics.run_by_user as executed_by,
+main.start_ts as start_ts,
+main.end_ts as end_ts,
+date(main.start_ts) as start_date
+from IDENTIFIER(:catalog || '.' || :schema || '.main' ) main
+inner join
+IDENTIFIER(:catalog || '.' || :schema || '.metrics' ) metrics
+on main.recon_table_id = metrics.recon_table_id
+where metrics.run_metrics.status = true
+order by metrics.inserted_ts desc, main.recon_id, main.target_table.table_name