remove ReconIntermediatePersist usage from prod code

Author: m-abulazm

src/databricks/labs/lakebridge/reconcile/compare.py

@@ -78,10 +78,6 @@ def reconcile_data(
         )
     )
 
-    # Write unmatched df to volume
-    df = ReconIntermediatePersist(spark, path).write_and_read_unmatched_df_with_volumes(df)
-    logger.debug(f"Unmatched data was written to {path} successfully")
-
     mismatch = _get_mismatch_data(df, source_alias, target_alias) if report_type in {"all", "data"} else None
 
     missing_in_src = (
@@ -450,9 +446,4 @@ def join_aggregate_data(
     joined_cols = source.columns + target.columns
     normalized_joined_cols = [DialectUtils.ansi_normalize_identifier(col) for col in joined_cols]
     joined_df = df.select(*normalized_joined_cols)
-
-    # Write the joined df to volume path
-    joined_volume_df = ReconIntermediatePersist(spark, path).write_and_read_unmatched_df_with_volumes(joined_df).cache()
-    logger.warning(f"Unmatched data is written to {path} successfully")
-
-    return joined_volume_df
+    return joined_df

src/databricks/labs/lakebridge/reconcile/trigger_recon_aggregate_service.py

@@ -67,13 +67,6 @@ def trigger_recon_aggregates(
                 recon_process_duration=recon_process_duration,
             )
 
-            (
-                ReconIntermediatePersist(
-                    spark=spark,
-                    path=utils.generate_volume_path(normalized_table_conf, reconcile_config.metadata_config),
-                ).clean_unmatched_df_from_volume()
-            )
-
         return TriggerReconService.verify_successful_reconciliation(
             generate_final_reconcile_aggregate_output(
                 recon_id=recon_capture.recon_id,

src/databricks/labs/lakebridge/reconcile/trigger_recon_service.py

@@ -232,10 +232,6 @@ def persist_delta_table(
             recon_process_duration=recon_process_duration,
             record_count=reconciler.get_record_count(table_conf, reconciler.report_type),
         )
-        if reconciler.report_type != "schema":
-            ReconIntermediatePersist(
-                spark=spark, path=utils.generate_volume_path(table_conf, reconcile_config.metadata_config)
-            ).clean_unmatched_df_from_volume()
 
     @staticmethod
     def verify_successful_reconciliation(