chore: add more tracking to resharding

pgdog-postgres-types/src/interface.rs

@@ -57,6 +57,15 @@ impl ToDataRowColumn for Option<i64> {
     }
 }
 
+impl ToDataRowColumn for Option<String> {
+    fn to_data_row_column(&self) -> Data {
+        match self {
+            Some(value) => ToDataRowColumn::to_data_row_column(value),
+            None => Data::null(),
+        }
+    }
+}
+
 impl ToDataRowColumn for usize {
     fn to_data_row_column(&self) -> Data {
         Bytes::copy_from_slice(self.to_string().as_bytes()).into()

pgdog-stats/src/resharding.rs

@@ -1,4 +1,4 @@
-use std::time::SystemTime;
+use std::{sync::Arc, time::SystemTime};
 
 use pgdog_config::ServerAuth;
 use serde::{Deserialize, Serialize};
@@ -13,7 +13,7 @@ pub struct TableCopy {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TableCopyState {
-    pub sql: String,
+    pub sql: Arc<String>,
     pub rows: usize,
     pub bytes: usize,
     pub bytes_per_sec: usize,
@@ -23,7 +23,7 @@ pub struct TableCopyState {
 impl Default for TableCopyState {
     fn default() -> Self {
         Self {
-            sql: String::default(),
+            sql: Arc::new(String::default()),
             rows: 0,
             bytes: 0,
             bytes_per_sec: 0,
@@ -67,16 +67,25 @@ pub struct Address {
 
 #[derive(Debug, Clone, PartialEq, Hash, Eq, Serialize, Deserialize)]
 pub struct SchemaStatement {
+    pub id: i64,
     pub user: User,
     pub shard: usize,
     pub sql: String,
     pub kind: StatementKind,
     pub sync_state: SyncState,
-    pub started_at: SystemTime,
+    pub started_at: Option<SystemTime>,
     pub table_schema: Option<String>,
     pub table_name: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Hash, Eq, Serialize, Deserialize)]
+pub struct SchemaStatementTask {
+    pub statement: SchemaStatement,
+    pub running: bool,
+    pub done: bool,
+    pub error: Option<String>,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum StatementKind {
     Table,

pgdog/src/admin/show_schema_sync.rs

@@ -38,26 +38,35 @@ impl Command for ShowSchemaSync {
         let mut messages = vec![rd.message()?];
         let now = SystemTime::now();
 
-        for entry in SchemaStatements::get().iter() {
-            let stmt = entry.key();
+        let mut entries: Vec<_> = SchemaStatements::get()
+            .iter()
+            .map(|task| task.clone())
+            .collect();
+        entries.sort_by_key(|k| if k.running { 0 } else { 1 });
 
-            let elapsed = now.duration_since(stmt.started_at).unwrap_or_default();
-            let elapsed_ms = elapsed.as_millis() as i64;
-            let elapsed_human = human_duration_display(elapsed);
+        for entry in entries {
+            let stmt = entry.statement;
+
+            let elapsed = stmt
+                .started_at
+                .map(|t| now.duration_since(t).unwrap_or_default());
+            let elapsed_ms: Option<i64> = elapsed.map(|e| e.as_millis() as i64);
+            let elapsed_human: Option<String> = elapsed.map(human_duration_display);
 
             let kind = stmt.kind.to_string();
             let sync_state = stmt.sync_state.to_string();
-            let started_at: DateTime<Local> = stmt.started_at.into();
-            let started_at = format_time(started_at);
+            let started_at: Option<String> = stmt
+                .started_at
+                .map(|started_at| format_time(DateTime::<Local>::from(started_at)));
 
             let mut row = DataRow::new();
             row.add(stmt.user.database.as_str())
                 .add(stmt.user.user.as_str())
                 .add(stmt.shard as i64)
                 .add(kind.as_str())
                 .add(sync_state.as_str())
-                .add(started_at.as_str())
-                .add(elapsed_human.as_str())
+                .add(started_at)
+                .add(elapsed_human)
                 .add(elapsed_ms)
                 .add(stmt.table_schema.as_deref().unwrap_or(""))
                 .add(stmt.table_name.as_deref().unwrap_or(""))

pgdog/src/backend/replication/logical/ee/mod.rs

@@ -1,7 +1,9 @@
 //! EE hooks.
 #![allow(dead_code, unused)]
 
-use pgdog_stats::Lsn;
+use pgdog_stats::{Lsn, SchemaStatementTask, TableCopyState};
+
+use crate::backend::replication::{logical::Error as LogicalError, status::TableCopy};
 
 use super::*;
 use std::time::Duration;
@@ -30,3 +32,11 @@ pub(crate) fn cutover_state(state: CutoverState) {
 }
 
 pub(crate) fn orchestrator_state(state: OrchestratorState) {}
+
+pub(crate) fn schema_sync_task(task: &SchemaStatementTask) {}
+
+pub(crate) fn data_sync_progress(table: &TableCopy, state: &TableCopyState) {}
+
+pub(crate) fn data_sync_done(table: &TableCopy) {}
+
+pub(crate) fn data_sync_error(table: &TableCopy, err: &LogicalError) {}

pgdog/src/backend/replication/logical/publisher/table.rs

@@ -230,7 +230,13 @@ impl Table {
                 },
 
                 result = copy_sub.copy_data(data_row) => {
-                    let (rows, bytes) = result?;
+                    let (rows, bytes) = match result {
+                        Ok(result) => result,
+                        Err(err) => {
+                            tracker.error(&err);
+                            return Err(err);
+                        }
+                    };
                     progress.update(copy_sub.bytes_sharded(), slot.lsn().lsn);
                     tracker.update_progress(bytes, rows);
                 }