diff --git a/terraform/environments/electronic-monitoring-data/dms_data_validation_glue_job_v2.tf b/terraform/environments/electronic-monitoring-data/dms_data_validation_glue_job_v2.tf
index 8d931646ff7..b6d452d332e 100644
--- a/terraform/environments/electronic-monitoring-data/dms_data_validation_glue_job_v2.tf
+++ b/terraform/environments/electronic-monitoring-data/dms_data_validation_glue_job_v2.tf
@@ -564,7 +564,7 @@ resource "aws_glue_job" "etl_rds_tbl_rows_hashvalue_to_s3_prq_yyyy_mm" {
   execution_property {
     max_concurrent_runs = 12
   }
-  
+
   default_arguments = {
     "--script_bucket_name"                  = module.s3-glue-job-script-bucket.bucket.id
     "--rds_db_host_ep"                      = split(":", aws_db_instance.database_2022.endpoint)[0]
@@ -815,31 +815,32 @@ resource "aws_glue_job" "dms_dv_on_rows_hashvalue_partitionby_yyyy_mm" {
   worker_type       = "G.2X"
   number_of_workers = 4
   default_arguments = {
-    "--script_bucket_name"               = module.s3-glue-job-script-bucket.bucket.id
-    "--rds_db_host_ep"                   = split(":", aws_db_instance.database_2022.endpoint)[0]
-    "--rds_db_pwd"                       = aws_db_instance.database_2022.password
-    "--rds_database_folder"              = ""
-    "--rds_db_schema_folder"             = "dbo"
-    "--rds_table_orignal_name"           = ""
-    "--table_pkey_column"                = ""
-    "--date_partition_column_name"       = ""
-    "--rds_hashed_rows_prq_parent_dir"   = "rds_tables_rows_hashed"
-    "--dms_prq_output_bucket"            = module.s3-dms-target-store-bucket.bucket.id
-    "--dms_prq_table_folder"             = ""
-    "--rds_only_where_clause"            = ""
-    "--prq_df_where_clause"              = ""
-    "--skip_columns_for_hashing"         = ""
-    "--rds_hashed_rows_prq_bucket"       = module.s3-dms-data-validation-bucket.bucket.id
-    "--glue_catalog_dv_bucket"           = module.s3-dms-data-validation-bucket.bucket.id
-    "--glue_catalog_db_name"             = aws_glue_catalog_database.dms_dv_glue_catalog_db.name
-    "--glue_catalog_tbl_name"            = "glue_df_output"
-    "--extra-py-files"                   = "s3://${module.s3-glue-job-script-bucket.bucket.id}/${aws_s3_object.aws_s3_object_pyzipfile_to_s3folder.id}"
-    "--continuous-log-logGroup"          = "/aws-glue/jobs/${aws_cloudwatch_log_group.dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.name}"
-    "--enable-continuous-cloudwatch-log" = "true"
-    "--enable-continuous-log-filter"     = "true"
-    "--enable-metrics"                   = "true"
-    "--enable-auto-scaling"              = "true"
-    "--conf"                             = <<EOF
+    "--script_bucket_name"                  = module.s3-glue-job-script-bucket.bucket.id
+    "--rds_db_host_ep"                      = split(":", aws_db_instance.database_2022.endpoint)[0]
+    "--rds_db_pwd"                          = aws_db_instance.database_2022.password
+    "--rds_database_folder"                 = ""
+    "--rds_db_schema_folder"                = "dbo"
+    "--rds_table_orignal_name"              = ""
+    "--table_pkey_column"                   = ""
+    "--date_partition_column_name"          = ""
+    "--rds_hashed_rows_prq_parent_dir"      = "rds_tables_rows_hashed"
+    "--dms_prq_output_bucket"               = module.s3-dms-target-store-bucket.bucket.id
+    "--dms_prq_table_folder"                = ""
+    "--rds_only_where_clause"               = ""
+    "--prq_df_where_clause"                 = ""
+    "--skip_columns_for_hashing"            = ""
+    "--read_rds_tbl_agg_stats_from_parquet" = "false"
+    "--rds_hashed_rows_prq_bucket"          = module.s3-dms-data-validation-bucket.bucket.id
+    "--glue_catalog_dv_bucket"              = module.s3-dms-data-validation-bucket.bucket.id
+    "--glue_catalog_db_name"                = aws_glue_catalog_database.dms_dv_glue_catalog_db.name
+    "--glue_catalog_tbl_name"               = "glue_df_output"
+    "--extra-py-files"                      = "s3://${module.s3-glue-job-script-bucket.bucket.id}/${aws_s3_object.aws_s3_object_pyzipfile_to_s3folder.id}"
+    "--continuous-log-logGroup"             = "/aws-glue/jobs/${aws_cloudwatch_log_group.dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.name}"
+    "--enable-continuous-cloudwatch-log"    = "true"
+    "--enable-continuous-log-filter"        = "true"
+    "--enable-metrics"                      = "true"
+    "--enable-auto-scaling"                 = "true"
+    "--conf"                                = <<EOF
 spark.sql.legacy.parquet.datetimeRebaseModeInRead=CORRECTED 
 --conf spark.sql.sources.partitionOverwriteMode=dynamic 
 --conf spark.sql.parquet.aggregatePushdown=true 
diff --git a/terraform/environments/electronic-monitoring-data/glue-job/dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.py b/terraform/environments/electronic-monitoring-data/glue-job/dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.py
index d02ed22e913..b57ef4a4ecd 100644
--- a/terraform/environments/electronic-monitoring-data/glue-job/dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.py
+++ b/terraform/environments/electronic-monitoring-data/glue-job/dms_dv_on_rows_hashvalue_partitionby_yyyy_mm.py
@@ -63,7 +63,8 @@
 OPTIONAL_INPUTS = [
     "rds_only_where_clause",
     "prq_df_where_clause",
-    "skip_columns_for_hashing"
+    "skip_columns_for_hashing",
+    "read_rds_tbl_agg_stats_from_parquet"
 ]
 
 AVAILABLE_ARGS_LIST = CustomPysparkMethods.resolve_args(DEFAULT_INPUTS_LIST+OPTIONAL_INPUTS)
@@ -216,12 +217,47 @@ def write_parquet_to_s3(df_dv_output: DataFrame, database, db_sch_tbl_name):
     LOGGER.info(f"""TABLE_PKEY_COLUMN = {TABLE_PKEY_COLUMN}""")
     LOGGER.info(f"""DATE_PARTITION_COLUMN_NAME = {DATE_PARTITION_COLUMN_NAME}""")
 
+    group_by_cols_list = ['year', 'month']
+    prq_df_where_clause = args.get("prq_df_where_clause", None)
+
     # EVALUATE RDS-DATAFRAME ROW-COUNT
-    rds_table_row_stats_df_agg = rds_jdbc_conn_obj.get_min_max_count_groupby_yyyy_mm(
-                                                    rds_table_orignal_name,
-                                                    DATE_PARTITION_COLUMN_NAME,
-                                                    TABLE_PKEY_COLUMN,
-                                                    args.get("rds_only_where_clause", None))
+    read_rds_tbl_agg_stats_from_parquet = args.get("read_rds_tbl_agg_stats_from_parquet", None)
+
+    if read_rds_tbl_agg_stats_from_parquet == 'true':
+        rds_table_row_stats_df_agg = CustomPysparkMethods.get_s3_parquet_df_v2(
+                                        f"""s3://{rds_hashed_rows_bucket_parent_dir}/rds_table_row_stats_df_agg""", 
+                                        CustomPysparkMethods.get_year_month_min_max_count_schema(TABLE_PKEY_COLUMN)
+                                        )
+        
+        if prq_df_where_clause is not None:
+            rds_table_row_stats_df_agg = rds_table_row_stats_df_agg.where(f"{prq_df_where_clause}")
+        # -----------------------------------------------------------------------------------------
+    else:
+        rds_table_row_stats_df_agg = rds_jdbc_conn_obj.get_min_max_count_groupby_yyyy_mm(
+                                                        rds_table_orignal_name,
+                                                        DATE_PARTITION_COLUMN_NAME,
+                                                        TABLE_PKEY_COLUMN,
+                                                        args.get("rds_only_where_clause", None))
+        
+        if S3Methods.check_s3_folder_path_if_exists(RDS_HASHED_ROWS_PRQ_BUCKET, 
+                                                    f"{rds_hashed_rows_bucket_parent_dir}/rds_table_row_stats_df_agg"):
+             prq_rds_table_row_stats_df_agg = CustomPysparkMethods.get_s3_parquet_df_v2(
+                                                f"""s3://{rds_hashed_rows_bucket_parent_dir}/rds_table_row_stats_df_agg""", 
+                                                rds_table_row_stats_df_agg.schema
+                                                )
+             prq_rds_table_row_stats_df_agg_updated = CustomPysparkMethods.update_df1_with_df2(prq_rds_table_row_stats_df_agg,
+                                                                            rds_table_row_stats_df_agg,
+                                                                            group_by_cols_list,
+                                                                            [e.name 
+                                                                                for e in rds_table_row_stats_df_agg.schema.fields
+                                                                                    if e.name not in group_by_cols_list
+                                                                            ]
+                                                      )
+             prq_rds_table_row_stats_df_agg_updated.write.mode("overwrite").parquet(f"""s3://{rds_hashed_rows_bucket_parent_dir}/rds_table_row_stats_df_agg""")
+        else:
+            rds_table_row_stats_df_agg.write.mode("overwrite").parquet(f"""s3://{rds_hashed_rows_bucket_parent_dir}/rds_table_row_stats_df_agg""")
+        # --------------------------------------------------------------------
+    # --------------------------------------------------------------------
     # +----+-----+-----------------+-----------------+-------------------+
     # |year|month|min_GPSPositionID|max_GPSPositionID|count_GPSPositionID|
     # +----+-----+-----------------+-----------------+-------------------+
@@ -265,21 +301,18 @@ def write_parquet_to_s3(df_dv_output: DataFrame, database, db_sch_tbl_name):
                                 )
 
 
-    group_by_cols_list = ['year', 'month']
-    prq_df_where_clause = args.get("prq_df_where_clause", None)
-
-
     if skipped_struct_fields_list:
         rds_hashed_rows_prq_df = CustomPysparkMethods.get_s3_parquet_df_v2(
                                     rds_hashed_rows_fulls3path, 
                                     CustomPysparkMethods.get_pyspark_hashed_table_schema(
-                                    TABLE_PKEY_COLUMN, skipped_struct_fields_list)
+                                                            TABLE_PKEY_COLUMN, 
+                                                            skipped_struct_fields_list)
                                     )
     else:
         rds_hashed_rows_prq_df = CustomPysparkMethods.get_s3_parquet_df_v2(
                                     rds_hashed_rows_fulls3path, 
                                     CustomPysparkMethods.get_pyspark_hashed_table_schema(
-                                    TABLE_PKEY_COLUMN)
+                                                            TABLE_PKEY_COLUMN)
                                     )
     
     if prq_df_where_clause is not None:
diff --git a/terraform/environments/electronic-monitoring-data/glue-job/reusable_module/glue_data_validation_lib.py b/terraform/environments/electronic-monitoring-data/glue-job/reusable_module/glue_data_validation_lib.py
index fadd9ae1f96..1f514ca840c 100644
--- a/terraform/environments/electronic-monitoring-data/glue-job/reusable_module/glue_data_validation_lib.py
+++ b/terraform/environments/electronic-monitoring-data/glue-job/reusable_module/glue_data_validation_lib.py
@@ -838,3 +838,41 @@ def get_pyspark_hashed_table_schema(in_pkey_column, sf_list=None):
             schema = schema.add(T.StructField("RowHash", T.StringType(), False))
 
             return schema
+
+    @staticmethod
+    def get_year_month_min_max_count_schema(in_pkey_column_str):
+
+        agg_schema = T.StructType([
+                T.StructField("year", T.IntegerType(), False),
+                T.StructField("month", T.IntegerType(), False),
+                T.StructField(f"min_{in_pkey_column_str}", T.LongType(), False),
+                T.StructField(f"max_{in_pkey_column_str}", T.LongType(), False),
+                T.StructField(f"count_{in_pkey_column_str}", T.LongType(), False)]
+                )
+        return agg_schema
+
+    @staticmethod
+    def update_df1_with_df2(df1: DataFrame, df2: DataFrame, 
+                            join_columns_list, 
+                            all_remaining_columns_list):
+        # Step 1: Find unmatched rows from df1 / df_parquet
+        df_unmatched_rows = df1.join(df2, join_columns_list, "left_anti")
+
+        # Step 2: Update matched rows between df2 / df_JDBC AND df1 / df_parquet
+        update_select_columns = [df2[c].alias(c) for c in all_remaining_columns_list]
+        df_updated_rows = df1.join(df2, join_columns_list, "inner") \
+            .select(
+                df1['year'],
+                df1['month'],
+                *update_select_columns
+            )
+
+        # Step 3: Include new rows from df2 / df_JDBC not in df1 / df_parquet
+        df_new_rows = df2.join(df1, join_columns_list, "left_anti")
+
+        # Step 4: Combine all type of rows in dataframes
+        final_df = df_unmatched_rows.union(df_updated_rows).union(df_new_rows)
+
+        final_df = final_df.orderBy(join_columns_list)
+
+        return final_df