Skip to content

feat: data-parity skill — TypeScript orchestrator, ClickHouse driver, partition support #493

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
suryaiyer95 wants to merge 21 commits into
base: main
Choose a base branch
from
+1,732 −57
Open

feat: data-parity skill — TypeScript orchestrator, ClickHouse driver, partition support #493

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
6d365ad
feat: add data-parity cross-database table comparison
suryaiyer95 Mar 27, 2026
44d7668
feat: add partition support to data_diff
suryaiyer95 Mar 27, 2026
e177f2d
feat: add categorical partition mode (string, enum, boolean)
suryaiyer95 Mar 27, 2026
d1cc932
fix: correct outcome shape handling in extractStats and formatOutcome
suryaiyer95 Mar 27, 2026
149066b
feat: rewrite data-parity skill with interactive, plan-first workflow
suryaiyer95 Mar 27, 2026
3caab30
fix: auto-discover extra_columns and exclude audit/timestamp columns …
aidtya Mar 28, 2026
550d431
fix: add `noLimit` option to driver `execute()` to prevent silent res…
aidtya Mar 28, 2026
f478bff
feat: detect auto-timestamp defaults from database catalog and confir…
aidtya Mar 28, 2026
b408017
fix: address code review findings in data-diff orchestrator
suryaiyer95 Mar 30, 2026
f2cee71
fix: address code review security and correctness findings
suryaiyer95 Mar 31, 2026
982316e
fix: resolve simulation suite failures — object stringification, erro…
suryaiyer95 Mar 31, 2026
05b6a02
refactor: remove existing-tool improvements — scope to data-diff only
suryaiyer95 Apr 1, 2026
6c60be1
refactor: revert .gitignore changes — scope to data-diff only
suryaiyer95 Apr 1, 2026
2c58580
fix: silence @clickhouse/client internal stderr logger to prevent TUI…
suryaiyer95 Apr 2, 2026
19c2376
fix: SQL injection hardening, target partition discovery, and local p…
suryaiyer95 Apr 2, 2026
7402408
feat: add Step 9 result presentation guidelines to data-parity skill
suryaiyer95 Apr 3, 2026
2caf381
fix: use correct outcome format for empty/fallback partition results
suryaiyer95 Apr 3, 2026
1bc67ef
chore: remove pack-local.ts — dev-only utility, not part of the feature
suryaiyer95 Apr 3, 2026
e41e5a0
feat: add data-parity skill to builder prompt with table and SQL quer…
suryaiyer95 Apr 3, 2026
b8147c9
fix: address code review findings — Oracle TRUNC, dialect-aware quoti…
suryaiyer95 Apr 3, 2026
997fc17
Merge branch 'main' into feat/data-parity-skill-improvements
suryaiyer95 Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
411 changes: 411 additions & 0 deletions .opencode/skills/data-parity/SKILL.md
View file
Open in desktop

Large diffs are not rendered by default.

8 changes: 4 additions & 4 deletions packages/drivers/src/bigquery.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* BigQuery driver using the `@google-cloud/bigquery` package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let BigQueryModule: any
Expand Down Expand Up @@ -37,8 +37,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
client = new BigQuery(options)
},

async execute(sql: string, limit?: number, binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, binds?: any[], execOptions?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = execOptions?.noLimit ? 0 : (limit ?? 1000)
const query = sql.replace(/;\s*$/, "")
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)

Expand All @@ -58,7 +58,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {

const [rows] = await client.query(options)
const columns = rows.length > 0 ? Object.keys(rows[0]) : []
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated ? rows.slice(0, effectiveLimit) : rows

return {
Expand Down
9 changes: 6 additions & 3 deletions packages/drivers/src/clickhouse.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
* Uses the official ClickHouse JS client which communicates over HTTP(S).
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let createClient: any
Expand Down Expand Up @@ -57,14 +57,17 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
clientConfig.clickhouse_settings = config.clickhouse_settings
}

// Silence the client's internal stderr logger — its ERROR-level output
// writes raw lines directly to stderr and corrupts terminal TUI rendering.
clientConfig.log = { level: 127 } // ClickHouseLogLevel.OFF = 127
client = createClient(clientConfig)
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
if (!client) {
throw new Error("ClickHouse client not connected — call connect() first")
}
const effectiveLimit = limit === undefined ? 1000 : limit
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql

// Strip string literals, then comments, for accurate SQL heuristic checks.
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/databricks.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* Databricks driver using the `@databricks/sql` package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let databricksModule: any
Expand Down Expand Up @@ -44,8 +44,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
})
},

async execute(sql: string, limit?: number, binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)
if (
Expand All @@ -65,7 +65,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
await operation.close()

const columns = rows.length > 0 ? Object.keys(rows[0]) : []
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated ? rows.slice(0, effectiveLimit) : rows

return {
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/duckdb.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* DuckDB driver using the `duckdb` package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let duckdb: any
Expand Down Expand Up @@ -105,8 +105,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
connection = db.connect()
},

async execute(sql: string, limit?: number, binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)

let finalSql = sql
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)
Expand All @@ -123,7 +123,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
: await query(finalSql)
const columns =
rows.length > 0 ? Object.keys(rows[0]) : []
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated ? rows.slice(0, effectiveLimit) : rows

return {
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/mysql.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* MySQL driver using the `mysql2` package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let mysql: any
Expand Down Expand Up @@ -41,8 +41,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
pool = mysql.createPool(poolConfig)
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)
if (
Expand All @@ -56,7 +56,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
const [rows, fields] = await pool.query(query)
const columns = fields?.map((f: any) => f.name) ?? []
const rowsArr = Array.isArray(rows) ? rows : []
const truncated = rowsArr.length > effectiveLimit
const truncated = effectiveLimit > 0 && rowsArr.length > effectiveLimit
const limitedRows = truncated
? rowsArr.slice(0, effectiveLimit)
: rowsArr
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/oracle.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* Oracle driver using the `oracledb` package (thin mode, pure JS).
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let oracledb: any
Expand Down Expand Up @@ -37,8 +37,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
})
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql
const isSelectLike = /^\s*(SELECT|WITH)\b/i.test(sql)

Expand All @@ -61,7 +61,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
const columns =
result.metaData?.map((m: any) => m.name) ??
(rows.length > 0 ? Object.keys(rows[0]) : [])
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated
? rows.slice(0, effectiveLimit)
: rows
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/postgres.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* PostgreSQL driver using the `pg` package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let pg: any
Expand Down Expand Up @@ -46,7 +46,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
pool = new Pool(poolConfig)
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const client = await pool.connect()
try {
if (config.statement_timeout) {
Expand All @@ -57,7 +57,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
}

let query = sql
const effectiveLimit = limit ?? 1000
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)
// Add LIMIT only for SELECT-like queries and if not already present
if (
Expand All @@ -70,7 +70,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {

const result = await client.query(query)
const columns = result.fields?.map((f: any) => f.name) ?? []
const truncated = result.rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && result.rows.length > effectiveLimit
const rows = truncated
? result.rows.slice(0, effectiveLimit)
: result.rows
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/redshift.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
* Uses svv_ system views for introspection.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let pg: any
Expand Down Expand Up @@ -46,10 +46,10 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
pool = new Pool(poolConfig)
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const client = await pool.connect()
try {
const effectiveLimit = limit ?? 1000
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql
const isSelectLike = /^\s*(SELECT|WITH|VALUES)\b/i.test(sql)
if (
Expand All @@ -62,7 +62,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {

const result = await client.query(query)
const columns = result.fields?.map((f: any) => f.name) ?? []
const truncated = result.rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && result.rows.length > effectiveLimit
const rows = truncated
? result.rows.slice(0, effectiveLimit)
: result.rows
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/snowflake.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
*/

import * as fs from "fs"
import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let snowflake: any
Expand Down Expand Up @@ -232,8 +232,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
})
},

async execute(sql: string, limit?: number, binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)
let query = sql
const isSelectLike = /^\s*(SELECT|WITH|VALUES|SHOW)\b/i.test(sql)
if (
Expand All @@ -245,7 +245,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
}

const result = await executeQuery(query, binds)
const truncated = result.rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && result.rows.length > effectiveLimit
const rows = truncated
? result.rows.slice(0, effectiveLimit)
: result.rows
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/sqlite.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
*/

import { Database } from "bun:sqlite"
import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
const dbPath = (config.path as string) ?? ":memory:"
Expand All @@ -22,9 +22,9 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
}
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
if (!db) throw new Error("SQLite connection not open")
const effectiveLimit = limit ?? 1000
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)

// Determine if this is a SELECT-like statement
const trimmed = sql.trim().toLowerCase()
Expand Down Expand Up @@ -60,7 +60,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
const stmt = db.prepare(query)
const rows = stmt.all() as any[]
const columns = rows.length > 0 ? Object.keys(rows[0]) : []
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated ? rows.slice(0, effectiveLimit) : rows

return {
Expand Down
8 changes: 4 additions & 4 deletions packages/drivers/src/sqlserver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
* SQL Server driver using the `mssql` (tedious) package.
*/

import type { ConnectionConfig, Connector, ConnectorResult, SchemaColumn } from "./types"
import type { ConnectionConfig, Connector, ConnectorResult, ExecuteOptions, SchemaColumn } from "./types"

export async function connect(config: ConnectionConfig): Promise<Connector> {
let mssql: any
Expand Down Expand Up @@ -42,8 +42,8 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
pool = await mssql.connect(mssqlConfig)
},

async execute(sql: string, limit?: number, _binds?: any[]): Promise<ConnectorResult> {
const effectiveLimit = limit ?? 1000
async execute(sql: string, limit?: number, _binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult> {
const effectiveLimit = options?.noLimit ? 0 : (limit ?? 1000)

let query = sql
const isSelectLike = /^\s*SELECT\b/i.test(sql)
Expand All @@ -69,7 +69,7 @@ export async function connect(config: ConnectionConfig): Promise<Connector> {
: (result.recordset?.columns
? Object.keys(result.recordset.columns)
: [])
const truncated = rows.length > effectiveLimit
const truncated = effectiveLimit > 0 && rows.length > effectiveLimit
const limitedRows = truncated ? rows.slice(0, effectiveLimit) : rows

return {
Expand Down
8 changes: 7 additions & 1 deletion packages/drivers/src/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,15 @@ export interface SchemaColumn {
nullable: boolean
}

export interface ExecuteOptions {
/** Skip the default LIMIT injection and post-truncation. Use when the caller
* needs the complete, untruncated result set (e.g. data-diff pipelines). */
noLimit?: boolean
}

export interface Connector {
connect(): Promise<void>
execute(sql: string, limit?: number, binds?: any[]): Promise<ConnectorResult>
execute(sql: string, limit?: number, binds?: any[], options?: ExecuteOptions): Promise<ConnectorResult>
listSchemas(): Promise<string[]>
listTables(schema: string): Promise<Array<{ name: string; type: string }>>
describeTable(schema: string, table: string): Promise<SchemaColumn[]>
Expand Down
Loading
Loading