Dashboard As Code for Reconcile

core/pom.xml

@@ -16,7 +16,7 @@
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
     <!-- 5.x starting from 5.0.0 requires JRE 11 -->
-    <mockito.version>4.11.0</mockito.version>
+    <mockito.version>5.11.0</mockito.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <slf4j.version>2.0.9</slf4j.version>
     <antlr.version>4.13.1</antlr.version>
@@ -47,7 +47,7 @@
     <dependency>
       <groupId>com.databricks</groupId>
       <artifactId>databricks-sdk-java</artifactId>
-      <version>0.21.0</version>
+      <version>0.24.1</version>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -78,7 +78,7 @@
     <dependency>
       <groupId>org.scalatest</groupId>
       <artifactId>scalatest_${scala.binary.version}</artifactId>
-      <version>3.3.0-SNAP3</version>
+      <version>3.3.0-SNAP4</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -230,4 +230,11 @@
       </plugin>
     </plugins>
   </build>
+    <repositories>
+        <repository>
+            <id>maven_central</id>
+            <name>Maven Central</name>
+            <url>https://repo.maven.apache.org/maven2/</url>
+        </repository>
+    </repositories>
 </project>

src/databricks/labs/remorph/resources/reconcile/dashboards/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/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/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/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/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