remove unused code after removing usage of ReconIntermediatePersist

Author: m-abulazm

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

@@ -1,13 +1,10 @@
 import logging
 from functools import reduce
-from pyspark.sql import DataFrame, SparkSession
+from pyspark.sql import DataFrame
 from pyspark.sql.functions import col, expr, lit
 
 from databricks.labs.lakebridge.reconcile.connectors.dialect_utils import DialectUtils
 from databricks.labs.lakebridge.reconcile.exception import ColumnMismatchException
-from databricks.labs.lakebridge.reconcile.recon_capture import (
-    ReconIntermediatePersist,
-)
 from databricks.labs.lakebridge.reconcile.recon_output_config import (
     DataReconcileOutput,
     MismatchOutput,
@@ -58,8 +55,6 @@ def reconcile_data(
     target: DataFrame,
     key_columns: list[str],
     report_type: str,
-    spark: SparkSession,
-    path: str,
 ) -> DataReconcileOutput:
     source_alias = "src"
     target_alias = "tgt"
@@ -414,13 +409,7 @@ def reconcile_agg_data_per_rule(
     return rule_reconcile_output
 
 
-def join_aggregate_data(
-    source: DataFrame,
-    target: DataFrame,
-    key_columns: list[str] | None,
-    spark: SparkSession,
-    path: str,
-) -> DataFrame:
+def join_aggregate_data(source: DataFrame, target: DataFrame, key_columns: list[str] | None) -> DataFrame:
     # TODO:  Integrate with reconcile_data function
 
     source_alias = "src"

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

@@ -7,7 +7,6 @@
     DatabaseConfig,
     ReconcileMetadataConfig,
 )
-from databricks.labs.lakebridge.reconcile import utils
 from databricks.labs.lakebridge.reconcile.compare import (
     capture_mismatch_data_and_columns,
     reconcile_data,
@@ -143,14 +142,11 @@ def _get_reconcile_output(
             options=table_conf.jdbc_reader_options,
         )
 
-        volume_path = utils.generate_volume_path(table_conf, self._metadata_config)
         return reconcile_data(
             source=src_data,
             target=tgt_data,
             key_columns=table_conf.join_columns,
             report_type=self._report_type,
-            spark=self._spark,
-            path=volume_path,
         )
 
     def _get_reconcile_aggregate_output(
@@ -230,8 +226,6 @@ def _get_reconcile_aggregate_output(
             self._target,
         ).build_queries()
 
-        volume_path = utils.generate_volume_path(table_conf, self._metadata_config)
-
         table_agg_output: list[AggregateQueryOutput] = []
 
         # Iterate over the grouped aggregates and reconcile the data
@@ -266,8 +260,6 @@ def _get_reconcile_aggregate_output(
                     source=src_data,
                     target=tgt_data,
                     key_columns=src_query_with_rules.group_by_columns,
-                    spark=self._spark,
-                    path=f"{volume_path}{src_query_with_rules.group_by_columns_as_str}",
                 )
             except DataSourceRuntimeException as e:
                 data_source_exception = e

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

@@ -4,11 +4,9 @@
 from databricks.sdk import WorkspaceClient
 
 from databricks.labs.lakebridge.config import ReconcileConfig, TableRecon
-from databricks.labs.lakebridge.reconcile import utils
 from databricks.labs.lakebridge.reconcile.exception import DataSourceRuntimeException, ReconciliationException
 from databricks.labs.lakebridge.reconcile.normalize_recon_config_service import NormalizeReconConfigService
 from databricks.labs.lakebridge.reconcile.recon_capture import (
-    ReconIntermediatePersist,
     generate_final_reconcile_aggregate_output,
 )
 from databricks.labs.lakebridge.reconcile.recon_config import AGG_RECONCILE_OPERATION_NAME

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

@@ -13,7 +13,6 @@
 from databricks.labs.lakebridge.reconcile.exception import DataSourceRuntimeException, ReconciliationException
 from databricks.labs.lakebridge.reconcile.recon_capture import (
     ReconCapture,
-    ReconIntermediatePersist,
     generate_final_reconcile_output,
 )
 from databricks.labs.lakebridge.reconcile.recon_config import Table, Schema
@@ -48,7 +47,7 @@ def trigger_recon(
         )
 
         for table_conf in table_recon.tables:
-            TriggerReconService.recon_one(spark, reconciler, recon_capture, reconcile_config, table_conf)
+            TriggerReconService.recon_one(reconciler, recon_capture, reconcile_config, table_conf)
 
         return TriggerReconService.verify_successful_reconciliation(
             generate_final_reconcile_output(
@@ -105,7 +104,6 @@ def create_recon_dependencies(
 
     @staticmethod
     def recon_one(
-        spark: SparkSession,
         reconciler: Reconciliation,
         recon_capture: ReconCapture,
         reconcile_config: ReconcileConfig,
@@ -119,15 +117,12 @@ def recon_one(
             reconciler, reconcile_config, normalized_table_conf
         )
 
-        TriggerReconService.persist_delta_table(
-            spark,
-            reconciler,
-            recon_capture,
-            schema_reconcile_output,
-            data_reconcile_output,
-            reconcile_config,
-            normalized_table_conf,
-            recon_process_duration,
+        recon_capture.start(
+            data_reconcile_output=data_reconcile_output,
+            schema_reconcile_output=schema_reconcile_output,
+            table_conf=table_conf,
+            recon_process_duration=recon_process_duration,
+            record_count=reconciler.get_record_count(table_conf, reconciler.report_type),
         )
 
         return schema_reconcile_output, data_reconcile_output
@@ -214,25 +209,6 @@ def _run_reconcile_data(
         except DataSourceRuntimeException as e:
             return DataReconcileOutput(exception=str(e))
 
-    @staticmethod
-    def persist_delta_table(
-        spark: SparkSession,
-        reconciler: Reconciliation,
-        recon_capture: ReconCapture,
-        schema_reconcile_output: SchemaReconcileOutput,
-        data_reconcile_output: DataReconcileOutput,
-        reconcile_config: ReconcileConfig,
-        table_conf: Table,
-        recon_process_duration: ReconcileProcessDuration,
-    ):
-        recon_capture.start(
-            data_reconcile_output=data_reconcile_output,
-            schema_reconcile_output=schema_reconcile_output,
-            table_conf=table_conf,
-            recon_process_duration=recon_process_duration,
-            record_count=reconciler.get_record_count(table_conf, reconciler.report_type),
-        )
-
     @staticmethod
     def verify_successful_reconciliation(
         reconcile_output: ReconcileOutput, operation_name: str = "reconcile"

tests/integration/reconcile/test_compare.py

@@ -44,8 +44,6 @@ def test_compare_data_for_report_all(
         target=target,
         key_columns=["s_suppkey", "s_nationkey"],
         report_type="all",
-        spark=mock_spark,
-        path=str(tmp_path),
     )
     expected = DataReconcileOutput(
         mismatch_count=1,
@@ -97,8 +95,6 @@ def test_compare_data_for_report_hash(mock_spark, tmp_path: Path):
         target=target,
         key_columns=["s_suppkey", "s_nationkey"],
         report_type="hash",
-        spark=mock_spark,
-        path=str(tmp_path),
     )
     expected = DataReconcileOutput(
         missing_in_src=missing_in_src,
@@ -281,8 +277,6 @@ def test_compare_data_special_column_names(mock_spark, tmp_path: Path):
         target=target,
         key_columns=["`s``supp#`", "`s_nation#`"],
         report_type="all",
-        spark=mock_spark,
-        path=str(tmp_path),
     )
     expected = DataReconcileOutput(
         mismatch_count=1,

tests/integration/reconcile/test_oracle_reconcile.py

@@ -76,7 +76,6 @@ def test_oracle_db_reconcile(mock_spark, mock_workspace_client, tmp_path):
     )
     with patch("databricks.labs.lakebridge.reconcile.utils.generate_volume_path", return_value=str(tmp_path)):
         _, data_reconcile_output = TriggerReconService.recon_one(
-            spark=mock_spark,
             reconciler=recon,
             recon_capture=recon_capture,
             reconcile_config=reconcile_config,