Merge pull request pavlospt#10 from pavlospt/chore/add-start-position…

…-flag

Add start_position argument to DiffPayload struct

README.md

@@ -73,6 +73,7 @@ Options:
       --only-sequences                        Only compare sequences, exclude data
       --only-count                            Do a quick test based on counts alone
       --chunk-size <CHUNK_SIZE>               The chunk size when comparing data [default: 10000]
+      --start-position <START_POSITION>       The start position for the comparison [default: 0]
       --max-connections <MAX_CONNECTIONS>     Max connections for Postgres pool [default: 100]
   -i, --include-tables [<INCLUDE_TABLES>...]  Tables included in the comparison
   -e, --exclude-tables [<EXCLUDE_TABLES>...]  Tables excluded from the comparison
@@ -104,6 +105,7 @@ async fn main() -> Result<()> {
     false, //only-sequences
     false, //only-count
     10_000, //chunk-size
+    0, //start-position
     100, //max-connections
     vec!["table1", "table2"], //include-tables (mutually exclusive with exclude-tables)
     vec!["table3", "table4"], //exclude-tables (mutually exclusive with include-tables)

examples/example_diff.rs

@@ -38,6 +38,9 @@ enum Commands {
         /// The chunk size when comparing data
         #[arg(long, default_value_t = 10000, required = false)]
         chunk_size: i64,
+        /// The start position for the comparison
+        #[arg(long, default_value_t = 0, required = false)]
+        start_position: i64,
         /// Max connections for Postgres pool
         #[arg(long, default_value_t = 100, required = false)]
         max_connections: i64,
@@ -70,6 +73,7 @@ async fn main() -> Result<()> {
             only_sequences,
             only_count,
             chunk_size,
+            start_position,
             max_connections,
             include_tables,
             exclude_tables,
@@ -82,6 +86,7 @@ async fn main() -> Result<()> {
                 *only_sequences,
                 *only_count,
                 *chunk_size,
+                *start_position,
                 *max_connections,
                 include_tables.to_vec(),
                 exclude_tables.to_vec(),

rust-pgdatadiff-client/src/main.rs

@@ -39,6 +39,9 @@ enum Commands {
         /// The chunk size when comparing data
         #[arg(long, default_value_t = 10000, required = false)]
         chunk_size: i64,
+        /// The start position for the comparison
+        #[arg(long, default_value_t = 0, required = false)]
+        start_position: i64,
         /// Max connections for Postgres pool
         #[arg(long, default_value_t = 100, required = false)]
         max_connections: i64,
@@ -69,6 +72,7 @@ async fn main_clap() -> Result<()> {
             only_sequences,
             only_count,
             chunk_size,
+            start_position,
             max_connections,
             include_tables,
             exclude_tables,
@@ -81,6 +85,7 @@ async fn main_clap() -> Result<()> {
                 *only_sequences,
                 *only_count,
                 *chunk_size,
+                *start_position,
                 *max_connections,
                 include_tables.to_vec(),
                 exclude_tables.to_vec(),
@@ -118,6 +123,10 @@ async fn main_inquire() -> Result<()> {
         .with_default("10000")
         .with_help_message("Enter the chunk size when comparing data")
         .prompt()?;
+    let start_position = Text::new("Start position for the comparison")
+        .with_default("0")
+        .with_help_message("Enter the start position for the comparison")
+        .prompt()?;
     let max_connections = Text::new("Number of DB connections to utilize")
         .with_default("100")
         .with_help_message("Enter the max connections for Postgres pool")
@@ -142,6 +151,7 @@ async fn main_inquire() -> Result<()> {
         only_sequences,
         only_count,
         chunk_size.parse::<i64>().unwrap(),
+        start_position.parse::<i64>().unwrap(),
         max_connections.parse::<i64>().unwrap(),
         include_tables
             .split_whitespace()

src/diff/diff_payload.rs

@@ -6,6 +6,7 @@ pub struct DiffPayload {
     only_sequences: bool,
     only_count: bool,
     chunk_size: i64,
+    start_position: i64,
     max_connections: i64,
     include_tables: Vec<String>,
     exclude_tables: Vec<String>,
@@ -23,6 +24,7 @@ impl DiffPayload {
     /// * `only_sequences` - A flag indicating whether to compare only sequences.
     /// * `count_only` - A flag indicating whether to count differences only.
     /// * `chunk_size` - The chunk size for processing large tables.
+    /// * `start_position` - The start position for the comparison.
     /// * `max_connections` - The maximum number of database connections to use.
     /// * `include_tables` - A list of tables to include in the comparison.
     /// * `exclude_tables` - A list of tables to exclude in the comparison.
@@ -39,6 +41,7 @@ impl DiffPayload {
         only_sequences: bool,
         only_count: bool,
         chunk_size: i64,
+        start_position: i64,
         max_connections: i64,
         include_tables: Vec<impl Into<String>>,
         exclude_tables: Vec<impl Into<String>>,
@@ -58,6 +61,7 @@ impl DiffPayload {
             only_sequences,
             only_count,
             chunk_size,
+            start_position,
             max_connections,
             include_tables: include_tables.into_iter().map(|t| t.into()).collect(),
             exclude_tables: exclude_tables.into_iter().map(|t| t.into()).collect(),
@@ -83,6 +87,9 @@ impl DiffPayload {
     pub fn chunk_size(&self) -> i64 {
         self.chunk_size
     }
+    pub fn start_position(&self) -> i64 {
+        self.start_position
+    }
     pub fn max_connections(&self) -> u32 {
         self.max_connections as u32
     }
@@ -111,6 +118,7 @@ mod tests {
             false,
             false,
             10000,
+            0,
             10,
             vec!["table1"],
             vec!["table2"],

src/diff/table/table_differ.rs

@@ -227,7 +227,7 @@ impl<TQE: TableSingleSourceQueryExecutor, DTQE: TableDualSourceQueryExecutor>
         start: Instant,
     ) -> Option<TableDiffOutput> {
         // Start data comparison
-        let mut position = 0;
+        let mut position = diff_payload.start_position();
         while position <= total_rows {
             let input = QueryHashDataInput::new(
                 schema_name.clone(),

src/diff/table/table_differ_tests.rs

@@ -30,6 +30,7 @@ mod tests {
             false,
             false,
             10000,
+            0,
             10,
             vec!["table1", "table2"],
             EMPTY_STRING_VEC,
@@ -74,6 +75,7 @@ mod tests {
             false,
             false,
             10000,
+            0,
             10,
             vec!["table1", "table2"],
             EMPTY_STRING_VEC,
@@ -138,6 +140,7 @@ mod tests {
             false,
             false,
             10000,
+            0,
             10,
             vec!["table1", "table2"],
             EMPTY_STRING_VEC,