Migrate parquet txn metadata processor to sdk

rust/processor/src/db/parquet/models/mod.rs

@@ -1,2 +1,3 @@
 pub mod default_models;
 pub mod fungible_asset_models;
+pub mod transaction_metadata_model;

rust/processor/src/db/parquet/models/transaction_metadata_model/mod.rs

@@ -0,0 +1,4 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod parquet_write_set_size_info;

rust/processor/src/db/postgres/models/transaction_metadata_model/mod.rs

@@ -4,6 +4,3 @@
 pub mod event_size_info;
 pub mod transaction_size_info;
 pub mod write_set_size_info;
-
-// parquet models
-pub mod parquet_write_set_size_info;

rust/processor/src/processors/parquet_processors/parquet_transaction_metadata_processor.rs

@@ -6,7 +6,7 @@ use crate::{
         create_parquet_handler_loop, generic_parquet_processor::ParquetDataGeneric,
         ParquetProcessingResult,
     },
-    db::postgres::models::transaction_metadata_model::parquet_write_set_size_info::WriteSetSize,
+    db::parquet::models::transaction_metadata_model::parquet_write_set_size_info::WriteSetSize,
     gap_detectors::ProcessingResult,
     processors::{parquet_processors::ParquetProcessorTrait, ProcessorName, ProcessorTrait},
     utils::{database::ArcDbPool, util::parse_timestamp},
@@ -92,31 +92,8 @@ impl ProcessorTrait for ParquetTransactionMetadataProcessor {
         let last_transaction_timestamp = transactions.last().unwrap().timestamp.clone();
         let mut transaction_version_to_struct_count: AHashMap<i64, i64> = AHashMap::new();
 
-        let mut write_set_sizes = vec![];
-
-        for txn in &transactions {
-            let txn_version = txn.version as i64;
-            let block_timestamp = parse_timestamp(txn.timestamp.as_ref().unwrap(), txn_version);
-            let size_info = match txn.size_info.as_ref() {
-                Some(size_info) => size_info,
-                None => {
-                    warn!(version = txn.version, "Transaction size info not found");
-                    continue;
-                },
-            };
-            for (index, write_set_size_info) in size_info.write_op_size_info.iter().enumerate() {
-                write_set_sizes.push(WriteSetSize::from_transaction_info(
-                    write_set_size_info,
-                    txn_version,
-                    index as i64,
-                    block_timestamp,
-                ));
-                transaction_version_to_struct_count
-                    .entry(txn_version)
-                    .and_modify(|e| *e += 1)
-                    .or_insert(1);
-            }
-        }
+        let write_set_sizes =
+            process_transaction(transactions, &mut transaction_version_to_struct_count);
 
         let write_set_size_info_parquet_data = ParquetDataGeneric {
             data: write_set_sizes,
@@ -143,3 +120,35 @@ impl ProcessorTrait for ParquetTransactionMetadataProcessor {
         &self.connection_pool
     }
 }
+
+pub fn process_transaction(
+    transactions: Vec<Transaction>,
+    transaction_version_to_struct_count: &mut AHashMap<i64, i64>,
+) -> Vec<WriteSetSize> {
+    let mut write_set_sizes = vec![];
+
+    for txn in transactions {
+        let txn_version = txn.version as i64;
+        let block_timestamp = parse_timestamp(txn.timestamp.as_ref().unwrap(), txn_version);
+        let size_info = match txn.size_info.as_ref() {
+            Some(size_info) => size_info,
+            None => {
+                warn!(version = txn.version, "Transaction size info not found");
+                continue;
+            },
+        };
+        for (index, write_set_size_info) in size_info.write_op_size_info.iter().enumerate() {
+            write_set_sizes.push(WriteSetSize::from_transaction_info(
+                write_set_size_info,
+                txn_version,
+                index as i64,
+                block_timestamp,
+            ));
+            transaction_version_to_struct_count
+                .entry(txn_version)
+                .and_modify(|e| *e += 1)
+                .or_insert(1);
+        }
+    }
+    write_set_sizes
+}

rust/processor/src/worker.rs

@@ -118,6 +118,9 @@ bitflags! {
 
         const FUNGIBLE_ASSET_ACTIVITIES = 1 << 26;
         const FUNGIBLE_ASSET_METADATA = 1 << 27;
+
+        // Transaction Metadata
+        const WRITE_SET_SIZE = 1 << 29;
     }
 }
 

rust/sdk-processor/src/config/indexer_processor_config.rs

@@ -6,6 +6,7 @@ use crate::{
     parquet_processors::{
         parquet_default_processor::ParquetDefaultProcessor,
         parquet_fungible_asset_processor::ParquetFungibleAssetProcessor,
+        parquet_transaction_metadata_processor::ParquetTransactionMetadataProcessor,
     },
     processors::{
         account_transactions_processor::AccountTransactionsProcessor, ans_processor::AnsProcessor,
@@ -89,6 +90,11 @@ impl RunnableConfig for IndexerProcessorConfig {
                     ParquetFungibleAssetProcessor::new(self.clone()).await?;
                 parquet_fungible_asset_processor.run_processor().await
             },
+            ProcessorConfig::ParquetTransactionMetadataProcessor(_) => {
+                let parquet_transaction_metadata_processor =
+                    ParquetTransactionMetadataProcessor::new(self.clone()).await?;
+                parquet_transaction_metadata_processor.run_processor().await
+            },
         }
     }
 

rust/sdk-processor/src/config/processor_config.rs

@@ -18,6 +18,7 @@ use processor::{
             parquet_write_set_changes::WriteSetChangeModel,
         },
         fungible_asset_models::parquet_v2_fungible_asset_activities::FungibleAssetActivity,
+        transaction_metadata_model::parquet_write_set_size_info::WriteSetSize,
     },
 };
 use serde::{Deserialize, Serialize};
@@ -70,6 +71,7 @@ pub enum ProcessorConfig {
     // ParquetProcessor
     ParquetDefaultProcessor(ParquetDefaultProcessorConfig),
     ParquetFungibleAssetProcessor(ParquetDefaultProcessorConfig),
+    ParquetTransactionMetadataProcessor(ParquetDefaultProcessorConfig),
 }
 
 impl ProcessorConfig {
@@ -86,6 +88,7 @@ impl ProcessorConfig {
     pub fn get_processor_status_table_names(&self) -> anyhow::Result<Vec<String>> {
         match self {
             ProcessorConfig::ParquetDefaultProcessor(config)
+            | ProcessorConfig::ParquetTransactionMetadataProcessor(config)
             | ProcessorConfig::ParquetFungibleAssetProcessor(config) => {
                 // Get the processor name as a prefix
                 let processor_name = self.name();
@@ -128,6 +131,9 @@ impl ProcessorConfig {
             ProcessorName::ParquetFungibleAssetProcessor => {
                 HashSet::from([FungibleAssetActivity::TABLE_NAME.to_string()])
             },
+            ProcessorName::ParquetTransactionMetadataProcessor => {
+                HashSet::from([WriteSetSize::TABLE_NAME.to_string()])
+            },
             _ => HashSet::new(), // Default case for unsupported processors
         }
     }

rust/sdk-processor/src/parquet_processors/mod.rs

@@ -30,6 +30,7 @@ use processor::{
             },
             parquet_v2_fungible_metadata::FungibleAssetMetadataModel,
         },
+        transaction_metadata_model::parquet_write_set_size_info::WriteSetSize,
     },
     worker::TableFlags,
 };
@@ -44,6 +45,7 @@ use strum::{Display, EnumIter};
 
 pub mod parquet_default_processor;
 pub mod parquet_fungible_asset_processor;
+pub mod parquet_transaction_metadata_processor;
 
 const GOOGLE_APPLICATION_CREDENTIALS: &str = "GOOGLE_APPLICATION_CREDENTIALS";
 
@@ -82,6 +84,8 @@ pub enum ParquetTypeEnum {
     FungibleAssetBalance,
     CurrentFungibleAssetBalance,
     CurrentUnifiedFungibleAssetBalance,
+    // txn metadata,
+    WriteSetSize,
 }
 
 /// Trait for handling various Parquet types.
@@ -155,6 +159,7 @@ impl_parquet_trait!(
     CurrentUnifiedFungibleAssetBalance,
     ParquetTypeEnum::CurrentUnifiedFungibleAssetBalance
 );
+impl_parquet_trait!(WriteSetSize, ParquetTypeEnum::WriteSetSize);
 
 #[derive(Debug, Clone)]
 #[enum_dispatch(ParquetTypeTrait)]
@@ -172,6 +177,7 @@ pub enum ParquetTypeStructs {
     FungibleAssetBalance(Vec<FungibleAssetBalance>),
     CurrentFungibleAssetBalance(Vec<CurrentFungibleAssetBalance>),
     CurrentUnifiedFungibleAssetBalance(Vec<CurrentUnifiedFungibleAssetBalance>),
+    WriteSetSize(Vec<WriteSetSize>),
 }
 
 impl ParquetTypeStructs {
@@ -202,6 +208,7 @@ impl ParquetTypeStructs {
             ParquetTypeEnum::CurrentUnifiedFungibleAssetBalance => {
                 ParquetTypeStructs::CurrentUnifiedFungibleAssetBalance(Vec::new())
             },
+            ParquetTypeEnum::WriteSetSize => ParquetTypeStructs::WriteSetSize(Vec::new()),
         }
     }
 
@@ -292,6 +299,12 @@ impl ParquetTypeStructs {
             ) => {
                 handle_append!(self_data, other_data)
             },
+            (
+                ParquetTypeStructs::WriteSetSize(self_data),
+                ParquetTypeStructs::WriteSetSize(other_data),
+            ) => {
+                handle_append!(self_data, other_data)
+            },
             _ => Err(ProcessorError::ProcessError {
                 message: "Mismatched buffer types in append operation".to_string(),
             }),