From 3d22e3b4dddddde7b1da14efd69fa6dbf21630a7 Mon Sep 17 00:00:00 2001 From: Yuun Lim <38443641+yuunlimm@users.noreply.github.com> Date: Thu, 21 Nov 2024 11:54:20 -0800 Subject: [PATCH] refactor parquet extractor backfill support function (#613) --- .../parquet_default_extractor.rs | 10 +--------- .../src/utils/parquet_extractor_helper.rs | 11 +++++------ 2 files changed, 6 insertions(+), 15 deletions(-) diff --git a/rust/sdk-processor/src/steps/parquet_default_processor/parquet_default_extractor.rs b/rust/sdk-processor/src/steps/parquet_default_processor/parquet_default_extractor.rs index 677bd8d43..58e46ce2e 100644 --- a/rust/sdk-processor/src/steps/parquet_default_processor/parquet_default_extractor.rs +++ b/rust/sdk-processor/src/steps/parquet_default_processor/parquet_default_extractor.rs @@ -99,15 +99,7 @@ impl Processable for ParquetDefaultExtractor { ]; // Populate the map based on opt-in tables - for (table_flag, enum_type, data) in data_types { - add_to_map_if_opted_in_for_backfill( - self.opt_in_tables, - &mut map, - table_flag, - enum_type, - data, - ); - } + add_to_map_if_opted_in_for_backfill(self.opt_in_tables, &mut map, data_types.to_vec()); Ok(Some(TransactionContext { data: map, diff --git a/rust/sdk-processor/src/utils/parquet_extractor_helper.rs b/rust/sdk-processor/src/utils/parquet_extractor_helper.rs index 7f6e95683..e6e52515f 100644 --- a/rust/sdk-processor/src/utils/parquet_extractor_helper.rs +++ b/rust/sdk-processor/src/utils/parquet_extractor_helper.rs @@ -6,12 +6,11 @@ use std::collections::HashMap; pub fn add_to_map_if_opted_in_for_backfill( opt_in_tables: TableFlags, map: &mut HashMap, - table_flag: TableFlags, - enum_type: ParquetTypeEnum, - data: ParquetTypeStructs, + data_types: Vec<(TableFlags, ParquetTypeEnum, ParquetTypeStructs)>, ) { - // Add to map if all tables are opted-in (empty) or if the specific flag is set - if opt_in_tables.is_empty() || opt_in_tables.contains(table_flag) { - map.insert(enum_type, data); + for (table_flag, enum_type, data) in data_types { + if opt_in_tables.is_empty() || opt_in_tables.contains(table_flag) { + map.insert(enum_type, data); + } } }