Skip to content

Commit f8374a4

Browse files
mason-sharpclaude
andcommitted
Batch concat_ws calls to support tables with 100+ columns
PostgreSQL limits functions to 100 arguments. The row-hashing expression used concat_ws('|', col1, ..., colN), which fails on wide tables. Split column expressions into groups of 99 with nested concat_ws calls. This affects both table-diff and merkle tree hash queries. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent eaf82b9 commit f8374a4

2 files changed

Lines changed: 85 additions & 1 deletion

File tree

db/queries/queries.go

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -555,6 +555,10 @@ func isNumericType(colType string) bool {
555555
// When allCols is provided, it returns concat_ws('|', col1_expr, col2_expr, ...)
556556
// with numeric/decimal columns wrapped in trim_scale() to normalize trailing zeros.
557557
// If allCols is nil/empty, falls back to the table-alias::text whole-row cast.
558+
//
559+
// PostgreSQL limits functions to 100 arguments. Since concat_ws uses 1 argument
560+
// for the separator, at most 99 column expressions fit per call. For wider
561+
// tables, the expressions are batched into nested concat_ws calls.
558562
func buildRowTextExpr(tableAlias string, allCols []string, colTypes map[string]string) string {
559563
if len(allCols) == 0 {
560564
return tableAlias + "::text"
@@ -570,7 +574,29 @@ func buildRowTextExpr(tableAlias string, allCols []string, colTypes map[string]s
570574
exprs[i] = fmt.Sprintf("COALESCE(%s::text, '')", qualifiedCol)
571575
}
572576
}
573-
return fmt.Sprintf("concat_ws('|', %s)", strings.Join(exprs, ", "))
577+
return concatWSBatched(exprs)
578+
}
579+
580+
// concatWSBatched produces a concat_ws('|', ...) expression. When len(exprs)
581+
// exceeds 99 (the max value-arguments per concat_ws call, since the separator
582+
// takes one slot), it splits the expressions into batches and nests the calls.
583+
func concatWSBatched(exprs []string) string {
584+
const maxArgs = 99 // 100 total - 1 for the separator
585+
586+
if len(exprs) <= maxArgs {
587+
return fmt.Sprintf("concat_ws('|', %s)", strings.Join(exprs, ", "))
588+
}
589+
590+
// Split into batches, wrap each in its own concat_ws, then combine.
591+
var batches []string
592+
for i := 0; i < len(exprs); i += maxArgs {
593+
end := i + maxArgs
594+
if end > len(exprs) {
595+
end = len(exprs)
596+
}
597+
batches = append(batches, fmt.Sprintf("concat_ws('|', %s)", strings.Join(exprs[i:end], ", ")))
598+
}
599+
return fmt.Sprintf("concat_ws('|', %s)", strings.Join(batches, ", "))
574600
}
575601

576602
func BlockHashSQL(schema, table string, primaryKeyCols []string, mode string, includeLower, includeUpper bool, filter string, allCols []string, colTypes map[string]string) (string, error) {

db/queries/queries_test.go

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
package queries
1313

1414
import (
15+
"fmt"
1516
"strings"
1617
"testing"
1718
)
@@ -409,3 +410,60 @@ func TestBlockHashSQL(t *testing.T) {
409410
})
410411
}
411412
}
413+
414+
func TestConcatWSBatched(t *testing.T) {
415+
t.Run("under limit is single concat_ws", func(t *testing.T) {
416+
exprs := make([]string, 50)
417+
for i := range exprs {
418+
exprs[i] = fmt.Sprintf("col%d", i)
419+
}
420+
result := concatWSBatched(exprs)
421+
// Should be a single concat_ws with all 50 expressions
422+
if strings.Count(result, "concat_ws(") != 1 {
423+
t.Errorf("expected 1 concat_ws call, got: %s", result)
424+
}
425+
})
426+
427+
t.Run("at limit of 99 is single concat_ws", func(t *testing.T) {
428+
exprs := make([]string, 99)
429+
for i := range exprs {
430+
exprs[i] = fmt.Sprintf("col%d", i)
431+
}
432+
result := concatWSBatched(exprs)
433+
if strings.Count(result, "concat_ws(") != 1 {
434+
t.Errorf("expected 1 concat_ws call for exactly 99 exprs, got: %s", result)
435+
}
436+
})
437+
438+
t.Run("100 expressions nests into batches", func(t *testing.T) {
439+
exprs := make([]string, 100)
440+
for i := range exprs {
441+
exprs[i] = fmt.Sprintf("col%d", i)
442+
}
443+
result := concatWSBatched(exprs)
444+
// Should have 3 concat_ws calls: 1 outer + 2 inner (99 + 1)
445+
if strings.Count(result, "concat_ws(") != 3 {
446+
t.Errorf("expected 3 concat_ws calls for 100 exprs, got %d in: %s",
447+
strings.Count(result, "concat_ws("), result)
448+
}
449+
// Every expression should be present
450+
for _, e := range exprs {
451+
if !strings.Contains(result, e) {
452+
t.Errorf("missing expression %s in result", e)
453+
}
454+
}
455+
})
456+
457+
t.Run("200 expressions nests into 3 inner batches", func(t *testing.T) {
458+
exprs := make([]string, 200)
459+
for i := range exprs {
460+
exprs[i] = fmt.Sprintf("col%d", i)
461+
}
462+
result := concatWSBatched(exprs)
463+
// 99 + 99 + 2 = 3 inner batches + 1 outer = 4 concat_ws calls
464+
if strings.Count(result, "concat_ws(") != 4 {
465+
t.Errorf("expected 4 concat_ws calls for 200 exprs, got %d in: %s",
466+
strings.Count(result, "concat_ws("), result)
467+
}
468+
})
469+
}

0 commit comments

Comments
 (0)