Add partition awareness to shard activity tracking

Author: WalBeh

README.md

@@ -314,6 +314,7 @@ This command helps identify which shards are receiving the most write activity b
 
 **Enhanced output features:**
 - **Checkpoint visibility**: Shows actual `local_checkpoint` values (CP Start → CP End → Delta)
+- **Partition awareness**: Separate tracking for partitioned tables (different partition_ident values)
 - **Activity trends**: 🔥 HOT (≥100/s), 📈 HIGH (≥50/s), 📊 MED (≥10/s), 📉 LOW (<10/s)
 - **Smart insights**: Identifies concentration patterns and load distribution (non-watch mode)
 - **Flexible filtering**: Exclude system tables, set minimum rates, hide replicas
@@ -328,16 +329,15 @@ This approach captures shards that become active during the observation period,
 
 Total checkpoint activity: 190,314 changes, Average rate: 2,109.0/sec
 
-   Rank | Schema.Table           | Shard | Node       | Type | CP Start  | CP End    | Checkpoint Δ | Rate/sec | Trend
-   ---------------------------------------------------------------------------------------------------------------
-   1    | gc.scheduled_jobs_log  | 0     | data-hot-8 | P    | 2,847,291 | 2,961,035 | 113,744      | 3,791.5  | 🔥 HOT
-   2    | TURVO.events           | 0     | data-hot-0 | P    | 1,234,567 | 1,280,404 | 45,837       | 1,527.9  | 🔥 HOT
-   3    | doc.user_actions       | 1     | data-hot-2 | P    | 987,654   | 1,018,387 | 30,733       | 1,024.4  | 🔥 HOT
+   Rank | Schema.Table           | Shard | Partition      | Node       | Type | Checkpoint Δ | Rate/sec | Trend
+   -----------------------------------------------------------------------------------------------------------
+   1    | gc.scheduled_jobs_log  | 0     | -              | data-hot-8 | P    | 113,744      | 3,791.5  | 🔥 HOT
+   2    | TURVO.events           | 0     | 04732dpl6osj8d | data-hot-0 | P    | 45,837       | 1,527.9  | 🔥 HOT
+   3    | doc.user_actions       | 1     | 04732dpk70rj6d | data-hot-2 | P    | 30,733       | 1,024.4  | 🔥 HOT
 
 Legend:
-  • CP Start: local_checkpoint at start of observation
-  • CP End: local_checkpoint at end of observation  
-  • Checkpoint Δ: CP End - CP Start (write operations during period)
+  • Checkpoint Δ: Write operations during observation period
+  • Partition: partition_ident (truncated if >14 chars, '-' if none)
 
 Insights:
   • 3 HOT shards (≥100 changes/sec) - consider load balancing
@@ -352,11 +352,11 @@ Insights:
 
 Total checkpoint activity: 190,314 changes, Average rate: 2,109.0/sec
 
-   Rank | Schema.Table           | Shard | Node       | Type | CP Start  | CP End    | Checkpoint Δ | Rate/sec | Trend
-   ---------------------------------------------------------------------------------------------------------------
-   1    | gc.scheduled_jobs_log  | 0     | data-hot-8 | P    | 2,847,291 | 2,961,035 | 113,744      | 3,791.5  | 🔥 HOT
-   2    | TURVO.events           | 0     | data-hot-0 | P    | 1,234,567 | 1,280,404 | 45,837       | 1,527.9  | 🔥 HOT
-   3    | doc.user_actions       | 1     | data-hot-2 | P    | 987,654   | 1,018,387 | 30,733       | 1,024.4  | 🔥 HOT
+   Rank | Schema.Table           | Shard | Partition      | Node       | Type | Checkpoint Δ | Rate/sec | Trend
+   -----------------------------------------------------------------------------------------------------------
+   1    | gc.scheduled_jobs_log  | 0     | -              | data-hot-8 | P    | 113,744      | 3,791.5  | 🔥 HOT
+   2    | TURVO.events           | 0     | 04732dpl6osj8d | data-hot-0 | P    | 45,837       | 1,527.9  | 🔥 HOT
+   3    | doc.user_actions       | 1     | 04732dpk70rj6d | data-hot-2 | P    | 30,733       | 1,024.4  | 🔥 HOT
 
 ━━━ Next update in 30s ━━━
 ```

src/xmover/analyzer.py

@@ -1082,7 +1082,7 @@ def format_activity_display(self, activities: List['ActiveShardActivity'],
             output.append("")
 
         # Create table headers
-        headers = ["Rank", "Schema.Table", "Shard", "Node", "Type", "CP Start", "CP End", "Checkpoint Δ", "Rate/sec", "Trend"]
+        headers = ["Rank", "Schema.Table", "Shard", "Partition", "Node", "Type", "Checkpoint Δ", "Rate/sec", "Trend"]
 
         # Calculate column widths
         col_widths = [len(h) for h in headers]
@@ -1094,10 +1094,9 @@ def format_activity_display(self, activities: List['ActiveShardActivity'],
             rank = str(i)
             table_id = activity.table_identifier
             shard_id = str(activity.shard_id)
+            partition = activity.partition_ident[:14] + "..." if len(activity.partition_ident) > 14 else activity.partition_ident or "-"
             node = activity.node_name
             shard_type = "P" if activity.is_primary else "R"
-            cp_start = f"{activity.snapshot1.local_checkpoint:,}"
-            cp_end = f"{activity.snapshot2.local_checkpoint:,}"
             checkpoint_delta = f"{activity.local_checkpoint_delta:,}"
             rate = f"{activity.activity_rate:.1f}" if activity.activity_rate >= 0.1 else "<0.1"
 
@@ -1111,7 +1110,7 @@ def format_activity_display(self, activities: List['ActiveShardActivity'],
             else:
                 trend = "📉 LOW"
 
-            row = [rank, table_id, shard_id, node, shard_type, cp_start, cp_end, checkpoint_delta, rate, trend]
+            row = [rank, table_id, shard_id, partition, node, shard_type, checkpoint_delta, rate, trend]
             rows.append(row)
 
             # Update column widths
@@ -1132,10 +1131,9 @@ def format_activity_display(self, activities: List['ActiveShardActivity'],
         if not watch_mode:
             output.append("")
             output.append("Legend:")
-            output.append("  • CP Start: local_checkpoint at start of observation")
-            output.append("  • CP End: local_checkpoint at end of observation")
-            output.append("  • Checkpoint Δ: CP End - CP Start (write operations during period)")
+            output.append("  • Checkpoint Δ: Write operations during observation period")
             output.append("  • Rate/sec: Checkpoint changes per second")
+            output.append("  • Partition: partition_ident (truncated if >14 chars, '-' if none)")
             output.append("  • Type: P=Primary, R=Replica")
             output.append("  • Trend: 🔥 HOT (≥100/s), 📈 HIGH (≥50/s), 📊 MED (≥10/s), 📉 LOW (<10/s)")
 

src/xmover/database.py

@@ -133,9 +133,10 @@ def translog_uncommitted_mb(self) -> float:
 
     @property
     def shard_identifier(self) -> str:
-        """Unique identifier for this shard"""
+        """Unique identifier for this shard including partition"""
         shard_type = "P" if self.is_primary else "R"
-        return f"{self.schema_name}.{self.table_name}:{self.shard_id}:{self.node_name}:{shard_type}"
+        partition = f":{self.partition_ident}" if self.partition_ident else ""
+        return f"{self.schema_name}.{self.table_name}:{self.shard_id}:{self.node_name}:{shard_type}{partition}"
 
 
 @dataclass

tests/test_active_shard_monitor.py

@@ -266,8 +266,7 @@ def test_format_activity_display_with_activities(self):
         assert "P" in display  # primary indicator
         assert "Legend:" in display
         assert "Trend:" in display  # new trend column explanation
-        assert "CP Start:" in display  # checkpoint start column
-        assert "CP End:" in display  # checkpoint end column
+        assert "Partition:" in display  # new partition column explanation
 
     def test_format_activity_display_empty(self):
         """Test formatting activity display with no data"""
@@ -382,6 +381,50 @@ def test_primary_replica_separation(self):
         # This test prevents the bug where we mixed primary CP End with replica CP Start
         # which created fake deltas like 129434 - 15876 = 113558
 
+    def test_partition_separation(self):
+        """Test that partitions within the same table/shard are tracked separately"""
+        # Create snapshots with same table/shard but different partitions
+        snapshot1 = [
+            # Partition 1
+            self.create_test_snapshot("TURVO", "appointmentFormFieldData_events", 0, "data-hot-8", True, 32684, 100.0),
+            # Partition 2 (same table/shard/node/type but different partition)
+            self.create_test_snapshot("TURVO", "appointmentFormFieldData_events", 0, "data-hot-8", True, 54289, 100.0),
+        ]
+        
+        # Modify partition_ident for the snapshots to simulate different partitions
+        snapshot1[0].partition_ident = "04732dpl6osj8d1g60o30c1g"
+        snapshot1[1].partition_ident = "04732dpl6os3adpm60o30c1g"
+        
+        snapshot2 = [
+            # Partition 1 progresses  
+            self.create_test_snapshot("TURVO", "appointmentFormFieldData_events", 0, "data-hot-8", True, 32800, 130.0),  # +116 delta
+            # Partition 2 progresses
+            self.create_test_snapshot("TURVO", "appointmentFormFieldData_events", 0, "data-hot-8", True, 54400, 130.0),  # +111 delta
+        ]
+        
+        # Set partition_ident for second snapshot
+        snapshot2[0].partition_ident = "04732dpl6osj8d1g60o30c1g"  
+        snapshot2[1].partition_ident = "04732dpl6os3adpm60o30c1g"
+        
+        activities = self.monitor.compare_snapshots(snapshot1, snapshot2, min_activity_threshold=1)
+        
+        # Should have 2 separate activities (partitions tracked separately)
+        assert len(activities) == 2
+        
+        # Verify deltas are calculated correctly for each partition
+        partition1_activity = next(a for a in activities if "04732dpl6osj8d1g60o30c1g" in a.snapshot1.shard_identifier)
+        partition2_activity = next(a for a in activities if "04732dpl6os3adpm60o30c1g" in a.snapshot1.shard_identifier)
+        
+        assert partition1_activity.local_checkpoint_delta == 116  # 32800 - 32684
+        assert partition2_activity.local_checkpoint_delta == 111  # 54400 - 54289
+        
+        # Verify they have different shard identifiers due to partition
+        assert partition1_activity.snapshot1.shard_identifier != partition2_activity.snapshot1.shard_identifier
+        assert ":04732dpl6osj8d1g60o30c1g" in partition1_activity.snapshot1.shard_identifier
+        assert ":04732dpl6os3adpm60o30c1g" in partition2_activity.snapshot1.shard_identifier
+        
+        # This test prevents mixing partitions which would create fake activity measurements
+    
     def test_format_activity_display_watch_mode(self):
         """Test that watch mode excludes legend and insights"""
         snapshot1 = self.create_test_snapshot("schema1", "table1", 1, "node1", True, 1000, 100.0)
@@ -404,13 +447,13 @@ def test_format_activity_display_watch_mode(self):
         normal_display = self.monitor.format_activity_display([activity], show_count=10, watch_mode=False)
         assert "Legend:" in normal_display
         assert "Insights:" in normal_display
-        assert "CP Start:" in normal_display
+        assert "Checkpoint Δ:" in normal_display
 
         # Test watch mode (should exclude legend and insights)
         watch_display = self.monitor.format_activity_display([activity], show_count=10, watch_mode=True)
         assert "Legend:" not in watch_display
         assert "Insights:" not in watch_display
-        assert "CP Start:" not in watch_display
+        assert "Checkpoint Δ" in watch_display  # Core data should still be present
 
         # But should still contain the core data
         assert "Most Active Shards" in watch_display