[parquet-sdk]migrate objects [pt2]

rust/processor/src/db/common/models/object_models/v2_objects.rs

@@ -19,6 +19,8 @@ use diesel_async::RunQueryDsl;
 use field_count::FieldCount;
 use serde::{Deserialize, Serialize};
 
+const DELETED_RESOURCE_OWNER_ADDRESS: &str = "Unknown";
+
 #[derive(Clone, Debug, Deserialize, FieldCount, Serialize)]
 pub struct RawObject {
     pub transaction_version: i64,
@@ -30,6 +32,7 @@ pub struct RawObject {
     pub allow_ungated_transfer: bool,
     pub is_deleted: bool,
     pub untransferrable: bool,
+    pub block_timestamp: chrono::NaiveDateTime,
 }
 
 pub trait ObjectConvertible {
@@ -46,6 +49,7 @@ pub struct RawCurrentObject {
     pub last_transaction_version: i64,
     pub is_deleted: bool,
     pub untransferrable: bool,
+    pub block_timestamp: chrono::NaiveDateTime,
 }
 
 pub trait CurrentObjectConvertible {
@@ -73,14 +77,10 @@ impl RawObject {
         txn_version: i64,
         write_set_change_index: i64,
         object_metadata_mapping: &ObjectAggregatedDataMapping,
+        block_timestamp: chrono::NaiveDateTime,
     ) -> anyhow::Result<Option<(Self, RawCurrentObject)>> {
         let address = standardize_address(&write_resource.address.to_string());
-        println!("address: {:?}", address);
         if let Some(object_aggregated_metadata) = object_metadata_mapping.get(&address) {
-            println!(
-                "object_aggregated_metadata: {:?}",
-                object_aggregated_metadata
-            );
             // do something
             let object_with_metadata = object_aggregated_metadata.object.clone();
             let object_core = object_with_metadata.object_core;
@@ -101,6 +101,7 @@ impl RawObject {
                     allow_ungated_transfer: object_core.allow_ungated_transfer,
                     is_deleted: false,
                     untransferrable,
+                    block_timestamp,
                 },
                 RawCurrentObject {
                     object_address: address,
@@ -111,6 +112,7 @@ impl RawObject {
                     last_transaction_version: txn_version,
                     is_deleted: false,
                     untransferrable,
+                    block_timestamp,
                 },
             )))
         } else {
@@ -126,7 +128,7 @@ impl RawObject {
         txn_version: i64,
         write_set_change_index: i64,
         object_mapping: &AHashMap<CurrentObjectPK, RawCurrentObject>,
-        conn: &mut DbPoolConnection<'_>,
+        conn: &mut Option<DbPoolConnection<'_>>,
         query_retries: u32,
         query_retry_delay_ms: u64,
     ) -> anyhow::Result<Option<(Self, RawCurrentObject)>> {
@@ -137,52 +139,91 @@ impl RawObject {
                 txn_version,
                 0, // Placeholder, this isn't used anyway
             );
-            let previous_object = if let Some(object) = object_mapping.get(&resource.address) {
-                object.clone()
+            // Add a logid here to handle None conn
+            if conn.is_none() {
+                // This is a hack to prevent the program for parquet
+                Ok(Some((
+                    Self {
+                        transaction_version: txn_version,
+                        write_set_change_index,
+                        object_address: resource.address.clone(),
+                        owner_address: DELETED_RESOURCE_OWNER_ADDRESS.to_string(),
+                        state_key_hash: resource.state_key_hash.clone(),
+                        guid_creation_num: BigDecimal::default(),
+                        allow_ungated_transfer: false,
+                        is_deleted: true,
+                        untransferrable: false,
+                        block_timestamp: chrono::NaiveDateTime::default(),
+                    },
+                    RawCurrentObject {
+                        object_address: resource.address.clone(),
+                        owner_address: DELETED_RESOURCE_OWNER_ADDRESS.to_string(),
+                        state_key_hash: resource.state_key_hash.clone(),
+                        last_guid_creation_num: BigDecimal::default(),
+                        allow_ungated_transfer: false,
+                        last_transaction_version: txn_version,
+                        is_deleted: true,
+                        untransferrable: false,
+                        block_timestamp: chrono::NaiveDateTime::default(),
+                    },
+                )))
             } else {
-                match Self::get_current_object(
-                    conn,
-                    &resource.address,
-                    query_retries,
-                    query_retry_delay_ms,
-                )
-                .await
-                {
-                    Ok(object) => object,
-                    Err(_) => {
-                        tracing::error!(
+                let previous_object = if let Some(object) = object_mapping.get(&resource.address) {
+                    object.clone()
+                } else if let Some(ref mut conn) = conn {
+                        match Self::get_current_object(
+                            conn,
+                            &resource.address,
+                            query_retries,
+                            query_retry_delay_ms,
+                        )
+                            .await
+                        {
+                            Ok(object) => object,
+                            Err(_) => {
+                                tracing::error!(
                             transaction_version = txn_version,
                             lookup_key = &resource.address,
                             "Missing current_object for object_address: {}. You probably should backfill db.",
                             resource.address,
                         );
+                                return Ok(None);
+                            },
+                        }
+                    } else {
+                        tracing::error!(
+                        transaction_version = txn_version,
+                        lookup_key = &resource.address,
+                        "Connection to DB is missing. You may need to investigate",
+                    );
                         return Ok(None);
+                    };
+                Ok(Some((
+                    Self {
+                        transaction_version: txn_version,
+                        write_set_change_index,
+                        object_address: resource.address.clone(),
+                        owner_address: previous_object.owner_address.clone(),
+                        state_key_hash: resource.state_key_hash.clone(),
+                        guid_creation_num: previous_object.last_guid_creation_num.clone(),
+                        allow_ungated_transfer: previous_object.allow_ungated_transfer,
+                        is_deleted: true,
+                        untransferrable: previous_object.untransferrable,
+                        block_timestamp: previous_object.block_timestamp,
                     },
-                }
-            };
-            Ok(Some((
-                Self {
-                    transaction_version: txn_version,
-                    write_set_change_index,
-                    object_address: resource.address.clone(),
-                    owner_address: previous_object.owner_address.clone(),
-                    state_key_hash: resource.state_key_hash.clone(),
-                    guid_creation_num: previous_object.last_guid_creation_num.clone(),
-                    allow_ungated_transfer: previous_object.allow_ungated_transfer,
-                    is_deleted: true,
-                    untransferrable: previous_object.untransferrable,
-                },
-                RawCurrentObject {
-                    object_address: resource.address,
-                    owner_address: previous_object.owner_address.clone(),
-                    state_key_hash: resource.state_key_hash,
-                    last_guid_creation_num: previous_object.last_guid_creation_num.clone(),
-                    allow_ungated_transfer: previous_object.allow_ungated_transfer,
-                    last_transaction_version: txn_version,
-                    is_deleted: true,
-                    untransferrable: previous_object.untransferrable,
-                },
-            )))
+                    RawCurrentObject {
+                        object_address: resource.address,
+                        owner_address: previous_object.owner_address.clone(),
+                        state_key_hash: resource.state_key_hash,
+                        last_guid_creation_num: previous_object.last_guid_creation_num.clone(),
+                        allow_ungated_transfer: previous_object.allow_ungated_transfer,
+                        last_transaction_version: txn_version,
+                        is_deleted: true,
+                        untransferrable: previous_object.untransferrable,
+                        block_timestamp: previous_object.block_timestamp,
+                    },
+                )))
+            }
         } else {
             Ok(None)
         }
@@ -210,6 +251,7 @@ impl RawObject {
                         last_transaction_version: res.last_transaction_version,
                         is_deleted: res.is_deleted,
                         untransferrable: res.untransferrable,
+                        block_timestamp: chrono::NaiveDateTime::default(), // this won't be used
                     });
                 },
                 Err(_) => {

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

@@ -6,3 +6,4 @@ pub mod fungible_asset_models;
 pub mod token_v2_models;
 pub mod transaction_metadata_model;
 pub mod user_transaction_models;
+pub mod object_models;

rust/processor/src/db/parquet/models/object_models/v2_objects.rs

@@ -5,49 +5,115 @@
 #![allow(clippy::extra_unused_lifetimes)]
 #![allow(clippy::unused_unit)]
 
-use super::v2_object_utils::{CurrentObjectPK, ObjectAggregatedDataMapping};
-use crate::{
-    db::postgres::models::default_models::move_resources::MoveResource,
-    schema::{current_objects, objects},
-    utils::{database::DbPoolConnection, util::standardize_address},
-};
-use ahash::AHashMap;
-use aptos_protos::transaction::v1::{DeleteResource, WriteResource};
-use bigdecimal::BigDecimal;
-use diesel::prelude::*;
-use diesel_async::RunQueryDsl;
+use bigdecimal::{ToPrimitive};
 use field_count::FieldCount;
 use serde::{Deserialize, Serialize};
+use allocative_derive::Allocative;
+use parquet_derive::ParquetRecordWriter;
+use crate::bq_analytics::generic_parquet_processor::NamedTable;
+use crate::bq_analytics::generic_parquet_processor::HasVersion;
+use crate::bq_analytics::generic_parquet_processor::GetTimeStamp;
+use crate::db::common::models::object_models::v2_objects::ObjectConvertible;
+use crate::db::common::models::object_models::v2_objects::CurrentObjectConvertible;
+use crate::db::common::models::object_models::v2_objects::RawObject;
+use crate::db::common::models::object_models::v2_objects::RawCurrentObject;
 
-#[derive(Clone, Debug, Deserialize, FieldCount, Identifiable, Insertable, Serialize)]
-pub struct RawObject {
-    pub transaction_version: i64,
+#[derive(
+    Allocative, Clone, Debug, Default, Deserialize, FieldCount, ParquetRecordWriter, Serialize,
+)]
+pub struct Object {
+    pub txn_version: i64,
     pub write_set_change_index: i64,
     pub object_address: String,
     pub owner_address: String,
     pub state_key_hash: String,
-    pub guid_creation_num: BigDecimal,
+    pub guid_creation_num: u64, //  BigDecimal,
     pub allow_ungated_transfer: bool,
     pub is_deleted: bool,
     pub untransferrable: bool,
+    #[allocative(skip)]
+    pub block_timestamp: chrono::NaiveDateTime,
 }
 
-pub trait ObjectConvertible {
-    fn from_raw(raw_item: RawObject) -> Self;
+impl NamedTable for Object {
+    const TABLE_NAME: &'static str = "objects";
 }
 
-#[derive(Clone, Debug, Deserialize, FieldCount, Identifiable, Insertable, Serialize)]
-pub struct RawCurrentObject {
+impl HasVersion for Object {
+    fn version(&self) -> i64 {
+        self.txn_version
+    }
+}
+
+impl GetTimeStamp for Object {
+    fn get_timestamp(&self) -> chrono::NaiveDateTime {
+        self.block_timestamp
+    }
+}
+
+impl ObjectConvertible for Object {
+    fn from_raw(raw_item: RawObject) -> Self {
+        Self {
+            txn_version: raw_item.transaction_version,
+            write_set_change_index: raw_item.write_set_change_index,
+            object_address: raw_item.object_address,
+            owner_address: raw_item.owner_address,
+            state_key_hash: raw_item.state_key_hash,
+            guid_creation_num: raw_item.guid_creation_num.to_u64().unwrap(),
+            allow_ungated_transfer: raw_item.allow_ungated_transfer,
+            is_deleted: raw_item.is_deleted,
+            untransferrable: raw_item.untransferrable,
+            block_timestamp: raw_item.block_timestamp,
+        }
+    }
+}
+
+#[derive(
+    Allocative, Clone, Debug, Default, Deserialize, FieldCount, ParquetRecordWriter, Serialize,
+)]
+pub struct CurrentObject {
     pub object_address: String,
     pub owner_address: String,
     pub state_key_hash: String,
     pub allow_ungated_transfer: bool,
-    pub last_guid_creation_num: BigDecimal,
+    pub last_guid_creation_num: u64, //  BigDecimal,
     pub last_transaction_version: i64,
     pub is_deleted: bool,
     pub untransferrable: bool,
+    #[allocative(skip)]
+    pub block_timestamp: chrono::NaiveDateTime,
+}
+
+
+impl NamedTable for CurrentObject {
+    const TABLE_NAME: &'static str = "objects";
+}
+
+impl HasVersion for CurrentObject {
+    fn version(&self) -> i64 {
+        self.last_transaction_version
+    }
 }
 
-pub trait CurrentObjectConvertible {
-    fn from_raw(raw_item: RawCurrentObject) -> Self;
+impl GetTimeStamp for CurrentObject {
+    fn get_timestamp(&self) -> chrono::NaiveDateTime {
+        self.block_timestamp
+    }
+}
+
+
+impl CurrentObjectConvertible for CurrentObject {
+    fn from_raw(raw_item: RawCurrentObject) -> Self {
+        Self {
+            object_address: raw_item.object_address,
+            owner_address: raw_item.owner_address,
+            state_key_hash: raw_item.state_key_hash,
+            allow_ungated_transfer: raw_item.allow_ungated_transfer,
+            last_guid_creation_num: raw_item.last_guid_creation_num.to_u64().unwrap(),
+            last_transaction_version: raw_item.last_transaction_version,
+            is_deleted: raw_item.is_deleted,
+            untransferrable: raw_item.untransferrable,
+            block_timestamp: raw_item.block_timestamp,
+        }
+    }
 }