[parquet-sdk][token_v2] migrate token ownership table [pt2] (#638)

### Description
- migrated token_ownership_v2
- 
### TestPlan
- Compared the number of rows from DE teams BigQuery table, testing BigQuery table, legacy processor table, and sdk processor table 

current_token_ownership_v2
![Screenshot 2024-12-11 at 7 57 05 PM](https://github.com/user-attachments/assets/30fe956c-b486-41b1-be15-e113395675e4)
BQ:
![Screenshot 2024-12-11 at 9 18 10 PM](https://github.com/user-attachments/assets/d09d98b3-93e6-4953-b691-db92e20ef62a)
PQ
![Screenshot 2024-12-11 at 7 54 00 PM](https://github.com/user-attachments/assets/725294b9-16c9-499f-9045-0b5c1ac6f0a3)

rust/processor/src/db/common/models/token_v2_models/raw_v2_token_ownerships.rs

@@ -257,7 +257,7 @@ impl RawTokenOwnershipV2 {
         prior_nft_ownership: &AHashMap<String, NFTOwnershipV2>,
         tokens_burned: &TokenV2Burned,
         object_metadatas: &ObjectAggregatedDataMapping,
-        conn: &mut DbPoolConnection<'_>,
+        conn: &mut Option<DbPoolConnection<'_>>,
         query_retries: u32,
         query_retry_delay_ms: u64,
     ) -> anyhow::Result<Option<(Self, RawCurrentTokenOwnershipV2)>> {
@@ -344,7 +344,7 @@ impl RawTokenOwnershipV2 {
         txn_timestamp: chrono::NaiveDateTime,
         prior_nft_ownership: &AHashMap<String, NFTOwnershipV2>,
         tokens_burned: &TokenV2Burned,
-        conn: &mut DbPoolConnection<'_>,
+        conn: &mut Option<DbPoolConnection<'_>>,
         query_retries: u32,
         query_retry_delay_ms: u64,
     ) -> anyhow::Result<Option<(Self, RawCurrentTokenOwnershipV2)>> {
@@ -370,7 +370,7 @@ impl RawTokenOwnershipV2 {
         txn_timestamp: chrono::NaiveDateTime,
         prior_nft_ownership: &AHashMap<String, NFTOwnershipV2>,
         tokens_burned: &TokenV2Burned,
-        conn: &mut DbPoolConnection<'_>,
+        conn: &mut Option<DbPoolConnection<'_>>,
         query_retries: u32,
         query_retry_delay_ms: u64,
     ) -> anyhow::Result<Option<(Self, RawCurrentTokenOwnershipV2)>> {
@@ -387,23 +387,36 @@ impl RawTokenOwnershipV2 {
                 match prior_nft_ownership.get(&token_address) {
                     Some(inner) => inner.owner_address.clone(),
                     None => {
-                        match CurrentTokenOwnershipV2Query::get_latest_owned_nft_by_token_data_id(
-                            conn,
-                            &token_address,
-                            query_retries,
-                            query_retry_delay_ms,
-                        )
-                        .await
-                        {
-                            Ok(nft) => nft.owner_address.clone(),
-                            Err(_) => {
+                        match conn {
+                            None => {
+                                // TODO: update message
                                 tracing::error!(
                                     transaction_version = txn_version,
                                     lookup_key = &token_address,
                                     "Failed to find current_token_ownership_v2 for burned token. You probably should backfill db."
                                 );
                                 DEFAULT_OWNER_ADDRESS.to_string()
                             },
+                            Some(ref mut conn) => {
+                                match CurrentTokenOwnershipV2Query::get_latest_owned_nft_by_token_data_id(
+                                    conn,
+                                    &token_address,
+                                    query_retries,
+                                    query_retry_delay_ms,
+                                )
+                                    .await
+                                {
+                                    Ok(nft) => nft.owner_address.clone(),
+                                    Err(_) => {
+                                        tracing::error!(
+                                    transaction_version = txn_version,
+                                    lookup_key = &token_address,
+                                    "Failed to find current_token_ownership_v2 for burned token. You probably should backfill db."
+                                );
+                                        DEFAULT_OWNER_ADDRESS.to_string()
+                                    },
+                                }
+                            }
                         }
                     },
                 }