From 430ff39a1a56b5c84a31111b6da8df6e973518e2 Mon Sep 17 00:00:00 2001
From: gtarraga <gt.tarraga@gmail.com>
Date: Wed, 31 Dec 2025 16:28:48 +0100
Subject: [PATCH 1/2] Fixed dashboard button

---
 README.md                                     | 4 ++++
 packages/big-daddy/src/dashboard/database.tsx | 6 ++++--
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index 0e09827..82f620f 100644
--- a/README.md
+++ b/README.md
@@ -67,6 +67,10 @@ packages/
 # Install dependencies
 pnpm install
 
+# Login to wrangler (needed to parse SQL w/ AI)
+# Or set a CLOUDFLARE_API_TOKEN in the projects .env file
+wrangler login
+
 # Run locally
 pnpm dev
 
diff --git a/packages/big-daddy/src/dashboard/database.tsx b/packages/big-daddy/src/dashboard/database.tsx
index 6b8fd82..ba27a11 100644
--- a/packages/big-daddy/src/dashboard/database.tsx
+++ b/packages/big-daddy/src/dashboard/database.tsx
@@ -38,6 +38,7 @@ export const DashboardPage = ({
 			</h2>
 			<form method="post" action={`/dash/${databaseId}/sql`} class="space-y-3">
 				<textarea
+					id="sql-query"
 					name="query"
 					placeholder="Enter SQL query..."
 					class="w-full px-3 py-2 border-2 border-black text-sm font-mono focus:outline-none focus:ring-2 focus:ring-offset-2 focus:ring-black"
@@ -48,12 +49,13 @@ export const DashboardPage = ({
 				<div class="flex gap-3">
 					<button
 						type="submit"
-						class="flex-1 bg-black text-white font-medium py-2 hover:bg-white hover:text-black hover:border-2 hover:border-black border-2 border-black transition"
+						class="flex-1 bg-black text-white font-medium py-2 border-2 border-black hover:bg-white hover:text-black transition"
 					>
 						Execute
 					</button>
 					<button
-						type="reset"
+						type="button"
+						onclick="document.getElementById('sql-query').value = ''"
 						class="flex-1 bg-white text-black font-medium py-2 border-2 border-black hover:bg-black hover:text-white transition"
 					>
 						Clear

From f3a0d6a6d94a8e0e9685d5871b179ba0ae049aa2 Mon Sep 17 00:00:00 2001
From: gtarraga <gt.tarraga@gmail.com>
Date: Thu, 1 Jan 2026 15:49:09 +0100
Subject: [PATCH 2/2] track per shard table row counts

---
 .../big-daddy/llm-resources/conductor-crud.md |   7 +-
 .../llm-resources/conductor-key-patterns.md   |  20 ++
 .../src/engine/async-jobs/reshard-table.ts    |  13 +
 .../src/engine/conductor/crud/delete.ts       |  22 ++
 .../src/engine/conductor/crud/insert.ts       |  45 ++-
 .../big-daddy/src/engine/topology/crud.ts     |  98 ++++++
 .../big-daddy/src/engine/topology/index.ts    |  59 +++-
 .../src/engine/topology/resharding.ts         |   5 +-
 .../big-daddy/src/engine/topology/tables.ts   |  20 ++
 .../big-daddy/src/engine/topology/types.ts    |  11 +
 .../big-daddy/test/engine/conductor.test.ts   |  38 +++
 .../big-daddy/test/engine/topology.test.ts    | 296 ++++++++++++++++++
 12 files changed, 626 insertions(+), 8 deletions(-)

diff --git a/packages/big-daddy/llm-resources/conductor-crud.md b/packages/big-daddy/llm-resources/conductor-crud.md
index a7e3338..768be79 100644
--- a/packages/big-daddy/llm-resources/conductor-crud.md
+++ b/packages/big-daddy/llm-resources/conductor-crud.md
@@ -20,12 +20,14 @@
 4. Execute per-shard INSERTs with remapped parameters
 5. Dispatch index maintenance events
 6. Invalidate cache
+7. **Bump per-shard row counts via `batchBumpTableShardRowCounts()`**
 
 **Key Functions:**
 - `groupInsertByShards()` - Groups rows by target shard, remaps placeholder indices
-- `extractInsertedRows()` - Builds row data for index maintenance
 
-**Important:** Multi-row INSERTs are split by shard key hash.
+**Important:** 
+- Multi-row INSERTs are split by shard key hash
+- Row counts use VALUES.length (not SQLite's rowsWritten which includes index writes)
 
 ---
 
@@ -50,6 +52,7 @@
 4. Execute on shards
 5. Dispatch index maintenance with old rows
 6. Invalidate cache
+7. **Decrement per-shard row counts via `batchBumpTableShardRowCounts()`**
 
 ---
 
diff --git a/packages/big-daddy/llm-resources/conductor-key-patterns.md b/packages/big-daddy/llm-resources/conductor-key-patterns.md
index 2604f2a..448f0d6 100644
--- a/packages/big-daddy/llm-resources/conductor-key-patterns.md
+++ b/packages/big-daddy/llm-resources/conductor-key-patterns.md
@@ -28,6 +28,26 @@ Every table has a hidden `_virtualShard` column:
   - INSERT: Specific key invalidation
   - UPDATE/DELETE: Full index invalidation (conservative)
 
+## Per-Shard Row Count Tracking
+
+Row counts are tracked per-shard in `table_shards.row_count`:
+
+| Operation | Action |
+|-----------|--------|
+| INSERT | Increment by VALUES.length per shard |
+| DELETE | Decrement by rowsAffected per shard |
+| UPDATE | No change (row count unchanged) |
+| Resharding | Set exact counts after atomic switch |
+
+**Why VALUES.length for INSERT?**
+SQLite's `rowsWritten` counts B-tree operations (table + indexes), which inflates the count 2x for tables with composite primary keys. Using VALUES.length gives the logical row count.
+
+**Topology Methods:**
+- `getTableShardRowCounts(table)` → get all shard counts
+- `bumpTableShardRowCount(table, shard, delta)` → single shard
+- `batchBumpTableShardRowCounts(table, deltaMap)` → multiple shards
+- `setTableShardRowCounts(table, countsMap)` → set exact (resharding)
+
 ## Effect Usage
 Table operations (`create-drop.ts`) use Effect for:
 - Typed errors (`TableAlreadyExistsError`, `TopologyFetchError`, etc.)
diff --git a/packages/big-daddy/src/engine/async-jobs/reshard-table.ts b/packages/big-daddy/src/engine/async-jobs/reshard-table.ts
index 95c34a2..40a8222 100644
--- a/packages/big-daddy/src/engine/async-jobs/reshard-table.ts
+++ b/packages/big-daddy/src/engine/async-jobs/reshard-table.ts
@@ -257,6 +257,19 @@ export async function processReshardTableJob(
 		await topologyStub.atomicStatusSwitch(tableName);
 		logger.info`Phase 4: Atomic status switch completed ${{ source }} ${{ table: tableName }}`;
 
+		// Phase 4B: Set row counts for new shards using the verified distribution
+		const rowCountsByShardId = new Map<number, number>();
+		for (const targetShardId of targetShardIds) {
+			// Use the breakdown from verification if available, otherwise use copy distribution
+			const rowCount =
+				verifyStats.verificationDetails?.targets.breakdown[targetShardId] ??
+				copyStats.distribution[targetShardId] ??
+				0;
+			rowCountsByShardId.set(targetShardId, rowCount);
+		}
+		await topologyStub.setTableShardRowCounts(tableName, rowCountsByShardId);
+		logger.info`Phase 4B: Row counts set for new shards ${{ source }} ${{ table: tableName }} ${{ rowCounts: Object.fromEntries(rowCountsByShardId) }}`;
+
 		// Phase 5: Delete old shard data
 		await deleteOldShardData(env, job.database_id, sourceShardId, tableName);
 		const shardId = sourceShardId;
diff --git a/packages/big-daddy/src/engine/conductor/crud/delete.ts b/packages/big-daddy/src/engine/conductor/crud/delete.ts
index 6f56593..038bce7 100644
--- a/packages/big-daddy/src/engine/conductor/crud/delete.ts
+++ b/packages/big-daddy/src/engine/conductor/crud/delete.ts
@@ -153,6 +153,28 @@ export async function handleDelete(
 		duration: 0,
 	}));
 
+	// STEP 8: Decrement row counts for each shard
+	if (result.rowsAffected && result.rowsAffected > 0) {
+		const { databaseId, topology } = context;
+		const topologyId = topology.idFromName(databaseId);
+		const topologyStub = topology.get(topologyId);
+
+		// Build delta map from per-shard results (negative for deletions)
+		const deltaByShard = new Map<number, number>();
+		for (let i = 0; i < shardsToQuery.length; i++) {
+			const shard = shardsToQuery[i];
+			const shardResult = results[i];
+			const rowsAffected = shardResult?.rowsAffected ?? 0;
+			if (shard && rowsAffected > 0) {
+				deltaByShard.set(shard.shard_id, -rowsAffected);
+			}
+		}
+
+		if (deltaByShard.size > 0) {
+			await topologyStub.batchBumpTableShardRowCounts(tableName, deltaByShard);
+		}
+	}
+
 	logger.info`DELETE query completed ${{ shardsQueried: shardsToQuery.length }} ${{ rowsAffected: result.rowsAffected }}`;
 
 	return result;
diff --git a/packages/big-daddy/src/engine/conductor/crud/insert.ts b/packages/big-daddy/src/engine/conductor/crud/insert.ts
index a4e5835..96ad476 100644
--- a/packages/big-daddy/src/engine/conductor/crud/insert.ts
+++ b/packages/big-daddy/src/engine/conductor/crud/insert.ts
@@ -193,7 +193,9 @@ export async function handleInsert(
 		params,
 	);
 
-	logger.info`Query plan determined for INSERT ${{ shardsSelected: planData.shardsToQuery.length }} ${{ indexesUsed: planData.virtualIndexes.length }} ${{ shardKey: planData.shardKey }}`;
+	logger.info`Query plan determined for INSERT ${{ shardsSelected: planData.shardsToQuery.length }} ${{
+		indexesUsed: planData.virtualIndexes.length,
+	}} ${{ shardKey: planData.shardKey }}`;
 
 	const allShards = planData.shardsToQuery;
 
@@ -214,12 +216,16 @@ export async function handleInsert(
 			? allShards
 			: allShards.filter((s) => perShardStatements.has(s.shard_id));
 
-	logger.info`INSERT rows grouped by shard ${{ shardsWithRows: perShardStatements.size }} ${{ totalShards: allShards.length }} ${{ shardsToQueryLength: shardsToQuery.length }}`;
+	logger.info`INSERT rows grouped by shard ${{ shardsWithRows: perShardStatements.size }} ${{ totalShards: allShards.length }} ${{
+		shardsToQueryLength: shardsToQuery.length,
+	}}`;
 
 	// STEP 3: Log write if resharding is in progress
 	await logWriteIfResharding(tableName, statement.type, query, params, context);
 
 	// STEP 4: Execute per-shard INSERTs
+	// Track per-shard results for row count bumping (declared here for scope)
+	const resultsPerShard = new Map<number, QueryResult>();
 	let execResult:
 		| {
 				results:
@@ -232,7 +238,6 @@ export async function handleInsert(
 		| undefined;
 	if (perShardStatements.size > 0) {
 		// Execute only the shards that have rows
-		const resultsPerShard = new Map<number, QueryResult>();
 		const shardStatsPerShard: ShardStats[] = [];
 
 		for (const [
@@ -358,6 +363,40 @@ export async function handleInsert(
 	// Add shard statistics
 	result.shardStats = execResult.shardStats;
 
+	// STEP 8: Bump row counts for each shard
+	// Use VALUES clause length (logical rows), not rowsAffected from SQLite
+	// SQLite's rowsWritten includes B-tree writes for indexes (e.g., composite PK),
+	// which would inflate counts by 2x for tables with indexes
+	const { databaseId, topology } = context;
+	const topologyId = topology.idFromName(databaseId);
+	const topologyStub = topology.get(topologyId);
+
+	if (perShardStatements.size > 0) {
+		// Build delta map from VALUES clause length per shard
+		const deltaByShard = new Map<number, number>();
+		for (const [shardId, { statement: shardStatement }] of perShardStatements) {
+			const rowsInserted = shardStatement.values?.length ?? 0;
+			if (rowsInserted > 0) {
+				deltaByShard.set(shardId, rowsInserted);
+			}
+		}
+
+		if (deltaByShard.size > 0) {
+			await topologyStub.batchBumpTableShardRowCounts(tableName, deltaByShard);
+		}
+	} else {
+		// Fallback case: shard key not in INSERT, rows were inserted on all shards
+		// Each shard gets all the rows from the original statement
+		const rowsInserted = statement.values?.length ?? 0;
+		if (rowsInserted > 0) {
+			const deltaByShard = new Map<number, number>();
+			for (const shard of shardsToQuery) {
+				deltaByShard.set(shard.shard_id, rowsInserted);
+			}
+			await topologyStub.batchBumpTableShardRowCounts(tableName, deltaByShard);
+		}
+	}
+
 	logger.info`INSERT query completed ${{ shardsQueried: shardsToQuery.length }} ${{ rowsAffected: result.rowsAffected }}`;
 
 	return result;
diff --git a/packages/big-daddy/src/engine/topology/crud.ts b/packages/big-daddy/src/engine/topology/crud.ts
index 0d81747..2a7785d 100644
--- a/packages/big-daddy/src/engine/topology/crud.ts
+++ b/packages/big-daddy/src/engine/topology/crud.ts
@@ -10,6 +10,7 @@
 import type {
 	AsyncJob,
 	ReshardingState,
+	ShardRowCount,
 	SqlParam,
 	StorageNode,
 	TableMetadata,
@@ -224,4 +225,101 @@ export class CRUDOperations {
 
 		return { success: true };
 	}
+
+	// ============================================================================
+	// Row Count Operations
+	// ============================================================================
+
+	/**
+	 * Get row counts for all shards of a table
+	 *
+	 * @param tableName - Name of the table
+	 * @returns Array of per-shard row counts
+	 */
+	getTableShardRowCounts(tableName: string): ShardRowCount[] {
+		return this.storage.sql
+			.exec(
+				`SELECT table_name, shard_id, row_count, updated_at 
+				 FROM table_shards 
+				 WHERE table_name = ? AND status = 'active'
+				 ORDER BY shard_id`,
+				tableName,
+			)
+			.toArray() as unknown as ShardRowCount[];
+	}
+
+	/**
+	 * Bump (increment or decrement) the row count for a specific shard
+	 * Clamps at 0 to prevent negative counts
+	 *
+	 * @param tableName - Name of the table
+	 * @param shardId - ID of the shard
+	 * @param delta - Amount to change (positive for inserts, negative for deletes)
+	 */
+	bumpTableShardRowCount(
+		tableName: string,
+		shardId: number,
+		delta: number,
+	): void {
+		const now = Date.now();
+		// Use MAX(0, ...) to clamp at 0
+		this.storage.sql.exec(
+			`UPDATE table_shards 
+			 SET row_count = MAX(0, row_count + ?), updated_at = ?
+			 WHERE table_name = ? AND shard_id = ?`,
+			delta,
+			now,
+			tableName,
+			shardId,
+		);
+	}
+
+	/**
+	 * Batch bump row counts for multiple shards of a table
+	 * More efficient than calling bumpTableShardRowCount multiple times
+	 *
+	 * @param tableName - Name of the table
+	 * @param deltaByShard - Map of shard_id to delta (amount to change)
+	 */
+	batchBumpTableShardRowCounts(
+		tableName: string,
+		deltaByShard: Map<number, number>,
+	): void {
+		const now = Date.now();
+		for (const [shardId, delta] of deltaByShard) {
+			this.storage.sql.exec(
+				`UPDATE table_shards 
+				 SET row_count = MAX(0, row_count + ?), updated_at = ?
+				 WHERE table_name = ? AND shard_id = ?`,
+				delta,
+				now,
+				tableName,
+				shardId,
+			);
+		}
+	}
+
+	/**
+	 * Set row counts for shards (used after resharding to set exact counts)
+	 *
+	 * @param tableName - Name of the table
+	 * @param countsByShardId - Map of shard_id to row_count
+	 */
+	setTableShardRowCounts(
+		tableName: string,
+		countsByShardId: Map<number, number>,
+	): void {
+		const now = Date.now();
+		for (const [shardId, rowCount] of countsByShardId) {
+			this.storage.sql.exec(
+				`UPDATE table_shards 
+				 SET row_count = ?, updated_at = ?
+				 WHERE table_name = ? AND shard_id = ?`,
+				rowCount,
+				now,
+				tableName,
+				shardId,
+			);
+		}
+	}
 }
diff --git a/packages/big-daddy/src/engine/topology/index.ts b/packages/big-daddy/src/engine/topology/index.ts
index 91b1ba6..f73171c 100644
--- a/packages/big-daddy/src/engine/topology/index.ts
+++ b/packages/big-daddy/src/engine/topology/index.ts
@@ -23,12 +23,13 @@ import type {
 import { AdministrationOperations } from "./administration";
 import { CRUDOperations } from "./crud";
 import { ReshardingOperations } from "./resharding";
-import { initializeSchemaTables } from "./tables";
+import { initializeSchemaTables, runSchemaMigrations } from "./tables";
 import type {
 	AsyncJob,
 	IndexStatus,
 	QueryPlanData,
 	ReshardingState,
+	ShardRowCount,
 	SqlParam,
 	StorageNode,
 	TableMetadata,
@@ -85,6 +86,14 @@ export class Topology extends DurableObject<Env> {
 	 */
 	private initializeSchema(): void {
 		initializeSchemaTables((sql: string) => this.ctx.storage.sql.exec(sql));
+		// Run migrations for existing schemas (adds new columns if missing)
+		runSchemaMigrations(
+			(sql: string) => this.ctx.storage.sql.exec(sql),
+			(tableName: string) =>
+				this.ctx.storage.sql
+					.exec(`PRAGMA table_info('${tableName}')`)
+					.toArray() as Array<{ name: string }>,
+		);
 	}
 
 	/**
@@ -143,6 +152,53 @@ export class Topology extends DurableObject<Env> {
 		return this.crud.updateTopology(updates);
 	}
 
+	// ============================================================================
+	// Row Count Operations
+	// ============================================================================
+
+	/**
+	 * Get row counts for all active shards of a table
+	 */
+	getTableShardRowCounts(tableName: string): ShardRowCount[] {
+		this.ensureCreated();
+		return this.crud.getTableShardRowCounts(tableName);
+	}
+
+	/**
+	 * Bump (increment or decrement) the row count for a specific shard
+	 * Clamps at 0 to prevent negative counts
+	 */
+	bumpTableShardRowCount(
+		tableName: string,
+		shardId: number,
+		delta: number,
+	): void {
+		this.ensureCreated();
+		this.crud.bumpTableShardRowCount(tableName, shardId, delta);
+	}
+
+	/**
+	 * Batch bump row counts for multiple shards of a table
+	 */
+	batchBumpTableShardRowCounts(
+		tableName: string,
+		deltaByShard: Map<number, number>,
+	): void {
+		this.ensureCreated();
+		this.crud.batchBumpTableShardRowCounts(tableName, deltaByShard);
+	}
+
+	/**
+	 * Set row counts for shards (used after resharding to set exact counts)
+	 */
+	setTableShardRowCounts(
+		tableName: string,
+		countsByShardId: Map<number, number>,
+	): void {
+		this.ensureCreated();
+		this.crud.setTableShardRowCounts(tableName, countsByShardId);
+	}
+
 	// ============================================================================
 	// Resharding Operations (from resharding.ts)
 	// ============================================================================
@@ -760,4 +816,5 @@ export type {
 	TopologyUpdates,
 	IndexStatus,
 	QueryPlanData,
+	ShardRowCount,
 };
diff --git a/packages/big-daddy/src/engine/topology/resharding.ts b/packages/big-daddy/src/engine/topology/resharding.ts
index c3a13a7..597ddde 100644
--- a/packages/big-daddy/src/engine/topology/resharding.ts
+++ b/packages/big-daddy/src/engine/topology/resharding.ts
@@ -82,8 +82,8 @@ export class ReshardingOperations {
 			const nodeId = nodes[nodeIndex]!.node_id;
 
 			this.storage.sql.exec(
-				`INSERT INTO table_shards (table_name, shard_id, node_id, status, created_at, updated_at)
-				 VALUES (?, ?, ?, 'pending', ?, ?)`,
+				`INSERT INTO table_shards (table_name, shard_id, node_id, status, row_count, created_at, updated_at)
+				 VALUES (?, ?, ?, 'pending', 0, ?, ?)`,
 				tableName,
 				shardId,
 				nodeId,
@@ -96,6 +96,7 @@ export class ReshardingOperations {
 				shard_id: shardId,
 				node_id: nodeId,
 				status: "pending",
+				row_count: 0,
 				created_at: now,
 				updated_at: now,
 			});
diff --git a/packages/big-daddy/src/engine/topology/tables.ts b/packages/big-daddy/src/engine/topology/tables.ts
index c02e1f2..c7e142b 100644
--- a/packages/big-daddy/src/engine/topology/tables.ts
+++ b/packages/big-daddy/src/engine/topology/tables.ts
@@ -5,6 +5,25 @@
  * These are executed during initialization to set up the persistent storage.
  */
 
+/**
+ * Run schema migrations that add new columns to existing tables.
+ * These are safe to run multiple times (idempotent).
+ */
+export function runSchemaMigrations(
+	sqlExec: (sql: string) => void,
+	tableInfo: (tableName: string) => Array<{ name: string }>,
+): void {
+	// Migration: Add row_count column to table_shards if not present
+	const tableShardsColumns = tableInfo("table_shards");
+	const hasRowCount = tableShardsColumns.some((col) => col.name === "row_count");
+	if (!hasRowCount && tableShardsColumns.length > 0) {
+		// Table exists but row_count column is missing - add it
+		sqlExec(
+			`ALTER TABLE table_shards ADD COLUMN row_count INTEGER NOT NULL DEFAULT 0`,
+		);
+	}
+}
+
 export function initializeSchemaTables(sqlExec: (sql: string) => void): void {
 	// Cluster metadata - stores if cluster is created
 	sqlExec(`
@@ -48,6 +67,7 @@ export function initializeSchemaTables(sqlExec: (sql: string) => void): void {
 			shard_id INTEGER NOT NULL,
 			node_id TEXT NOT NULL,
 			status TEXT NOT NULL DEFAULT 'active' CHECK(status IN ('active', 'pending', 'to_be_deleted', 'failed')),
+			row_count INTEGER NOT NULL DEFAULT 0,
 			created_at INTEGER NOT NULL,
 			updated_at INTEGER NOT NULL,
 			PRIMARY KEY (table_name, shard_id),
diff --git a/packages/big-daddy/src/engine/topology/types.ts b/packages/big-daddy/src/engine/topology/types.ts
index 2ec2892..79f4619 100644
--- a/packages/big-daddy/src/engine/topology/types.ts
+++ b/packages/big-daddy/src/engine/topology/types.ts
@@ -33,10 +33,21 @@ export interface TableShard {
 	shard_id: number;
 	node_id: string;
 	status: "active" | "pending" | "to_be_deleted" | "failed";
+	row_count: number;
 	created_at: number;
 	updated_at: number;
 }
 
+/**
+ * Per-shard row count for a table
+ */
+export interface ShardRowCount {
+	table_name: string;
+	shard_id: number;
+	row_count: number;
+	updated_at: number;
+}
+
 /**
  * Resharding state - tracks ongoing resharding operations
  * Allows the conductor to know when to log changes and when resharding is complete
diff --git a/packages/big-daddy/test/engine/conductor.test.ts b/packages/big-daddy/test/engine/conductor.test.ts
index 19fbe91..f916233 100644
--- a/packages/big-daddy/test/engine/conductor.test.ts
+++ b/packages/big-daddy/test/engine/conductor.test.ts
@@ -179,6 +179,44 @@ describe("Conductor", () => {
 		expect(remaining.rows[0]).toMatchObject({ id: 200, name: "Bob" });
 	});
 
+	it("should maintain per-shard row_count on INSERT and DELETE", async () => {
+		const dbId = "test-rowcount-maintenance";
+		const sql = await createConnection(dbId, { nodes: 3 }, env);
+
+		await sql`CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)`;
+
+		await sql`INSERT INTO users (id, name) VALUES (${1}, ${"Alice"})`;
+		await sql`INSERT INTO users (id, name) VALUES (${2}, ${"Bob"})`;
+		await sql`INSERT INTO users (id, name) VALUES (${3}, ${"Charlie"})`;
+
+		const topologyStub = env.TOPOLOGY.get(env.TOPOLOGY.idFromName(dbId));
+		const topology1 = await topologyStub.getTopology();
+		const userShards1 = topology1.table_shards.filter(
+			(s: TableShard) => s.table_name === "users",
+		);
+		const sum1 = userShards1.reduce(
+			(acc: number, s: TableShard) => acc + s.row_count,
+			0,
+		);
+		const countResult1 = await sql`SELECT COUNT(*) as cnt FROM users`;
+		const count1 = countResult1.rows[0]?.cnt;
+		expect(sum1).toBe(count1);
+
+		await sql`DELETE FROM users WHERE id = ${2}`;
+
+		const topology2 = await topologyStub.getTopology();
+		const userShards2 = topology2.table_shards.filter(
+			(s: TableShard) => s.table_name === "users",
+		);
+		const sum2 = userShards2.reduce(
+			(acc: number, s: TableShard) => acc + s.row_count,
+			0,
+		);
+		const countResult2 = await sql`SELECT COUNT(*) as cnt FROM users`;
+		const count2 = countResult2.rows[0]?.cnt;
+		expect(sum2).toBe(count2);
+	});
+
 	it("should route SELECT to correct shard when WHERE filters on shard key", async () => {
 		const dbId = "test-select-routing";
 		const sql = await createConnection(dbId, { nodes: 2 }, env);
diff --git a/packages/big-daddy/test/engine/topology.test.ts b/packages/big-daddy/test/engine/topology.test.ts
index 4017541..9b010a2 100644
--- a/packages/big-daddy/test/engine/topology.test.ts
+++ b/packages/big-daddy/test/engine/topology.test.ts
@@ -2,6 +2,7 @@ import { env } from "cloudflare:test";
 import { parse } from "@databases/sqlite-ast";
 import { describe, expect, it } from "vitest";
 import type {
+	ShardRowCount,
 	StorageNode,
 	TableShard,
 	VirtualIndex,
@@ -726,6 +727,301 @@ describe("Topology Durable Object", () => {
 		});
 	});
 
+	describe("Row Count Operations", () => {
+		it("should initialize shards with row_count = 0", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-init-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 3,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			const topology = await stub.getTopology();
+			const userShards = topology.table_shards.filter(
+				(s: TableShard) => s.table_name === "users",
+			);
+
+			expect(userShards).toHaveLength(3);
+			userShards.forEach((shard: TableShard) => {
+				expect(shard.row_count).toBe(0);
+			});
+		});
+
+		it("should get row counts for a table", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-get-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 2,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			expect(rowCounts).toHaveLength(2);
+			rowCounts.forEach((rc: ShardRowCount) => {
+				expect(rc.table_name).toBe("users");
+				expect(rc.row_count).toBe(0);
+				expect(rc.updated_at).toBeGreaterThan(0);
+			});
+		});
+
+		it("should bump row count for a shard (positive delta)", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-bump-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 2,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// Bump shard 0 by +10
+			await stub.bumpTableShardRowCount("users", 0, 10);
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			const shard0 = rowCounts.find((rc: ShardRowCount) => rc.shard_id === 0);
+			const shard1 = rowCounts.find((rc: ShardRowCount) => rc.shard_id === 1);
+
+			expect(shard0?.row_count).toBe(10);
+			expect(shard1?.row_count).toBe(0);
+		});
+
+		it("should bump row count for a shard (negative delta)", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-bump-2");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 2,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// First add 100 rows
+			await stub.bumpTableShardRowCount("users", 0, 100);
+			// Then remove 30 rows
+			await stub.bumpTableShardRowCount("users", 0, -30);
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			const shard0 = rowCounts.find((rc: ShardRowCount) => rc.shard_id === 0);
+
+			expect(shard0?.row_count).toBe(70);
+		});
+
+		it("should clamp row count at 0 (prevent negative)", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-clamp-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 2,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// Try to decrement below 0
+			await stub.bumpTableShardRowCount("users", 0, -100);
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			const shard0 = rowCounts.find((rc: ShardRowCount) => rc.shard_id === 0);
+
+			expect(shard0?.row_count).toBe(0); // Clamped at 0
+		});
+
+		it("should batch bump row counts for multiple shards", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-batch-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(3);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 3,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// Batch bump: shard 0 +10, shard 1 +20, shard 2 +30
+			const deltaByShard = new Map<number, number>([
+				[0, 10],
+				[1, 20],
+				[2, 30],
+			]);
+			await stub.batchBumpTableShardRowCounts("users", deltaByShard);
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			expect(rowCounts.find((rc: ShardRowCount) => rc.shard_id === 0)?.row_count).toBe(10);
+			expect(rowCounts.find((rc: ShardRowCount) => rc.shard_id === 1)?.row_count).toBe(20);
+			expect(rowCounts.find((rc: ShardRowCount) => rc.shard_id === 2)?.row_count).toBe(30);
+		});
+
+		it("should set row counts for shards (used after resharding)", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-set-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 2,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// Set exact row counts
+			const countsByShardId = new Map<number, number>([
+				[0, 500],
+				[1, 750],
+			]);
+			await stub.setTableShardRowCounts("users", countsByShardId);
+
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			expect(rowCounts.find((rc: ShardRowCount) => rc.shard_id === 0)?.row_count).toBe(500);
+			expect(rowCounts.find((rc: ShardRowCount) => rc.shard_id === 1)?.row_count).toBe(750);
+		});
+
+		it("should only return row counts for active shards", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-active-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(3);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 1,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			// Set row count on shard 0
+			await stub.bumpTableShardRowCount("users", 0, 100);
+
+			// Create pending shards (resharding)
+			const changeLogId = "test-log";
+			await stub.createPendingShards("users", 2, changeLogId);
+
+			// getTableShardRowCounts should only return active shards (not pending)
+			const rowCounts = await stub.getTableShardRowCounts("users");
+			expect(rowCounts).toHaveLength(1);
+			expect(rowCounts[0]?.shard_id).toBe(0);
+			expect(rowCounts[0]?.row_count).toBe(100);
+		});
+
+		it("should update updated_at timestamp on row count changes", async () => {
+			const id = env.TOPOLOGY.idFromName("test-rowcount-timestamp-1");
+			const stub = env.TOPOLOGY.get(id);
+
+			await stub.create(2);
+			await stub.updateTopology({
+				tables: {
+					add: [
+						{
+							table_name: "users",
+							primary_key: "id",
+							primary_key_type: "INTEGER",
+							shard_strategy: "hash",
+							shard_key: "id",
+							num_shards: 1,
+							block_size: 500,
+						},
+					],
+				},
+			});
+
+			const initialRowCounts = await stub.getTableShardRowCounts("users");
+			const initialTimestamp = initialRowCounts[0]?.updated_at ?? 0;
+
+			// Wait a bit and bump count
+			await new Promise((resolve) => setTimeout(resolve, 10));
+			await stub.bumpTableShardRowCount("users", 0, 5);
+
+			const updatedRowCounts = await stub.getTableShardRowCounts("users");
+			const updatedTimestamp = updatedRowCounts[0]?.updated_at ?? 0;
+
+			expect(updatedTimestamp).toBeGreaterThan(initialTimestamp);
+		});
+	});
+
 	describe("getQueryPlanData", () => {
 		describe("SELECT queries", () => {
 			it("should return all shards when WHERE clause does not match shard key or indexes", async () => {