From 9cd535fb6a6976e4a3ba189b630714afa732e5d5 Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 4 Jun 2026 21:11:23 +0400 Subject: [PATCH] perf(bulker) snowflake: sort dedup CTAS by timestamp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dedup CTAS used to emit rows in window-function-output order (roughly PK-grouped), which meant INSERT later wrote new T micro-partitions whose TO_DATE(ts) ranges spanned the whole batch. T is clustered by TO_DATE(timestamp) (sfAlterClusteringKeyTemplate), so those wide new micro-partitions force auto-clustering to re-sort them later — billable warehouse work. Add ORDER BY {ts} to the dedup template (gated on a new DedupOrderBy QueryPayload field) so the dedup rows hit storage in timestamp order. INSERT then writes new micro-partitions whose TO_DATE(ts) min/max are already tight along the cluster key; auto-clustering has almost nothing to do. UPDATE is unaffected (rewriting micro-partitions preserves T's existing clustering); the join cost is unaffected (PK-keyed, not ts-keyed). ORDER BY adds an O(N log N) sort on the already-deduped row set — sub-second for typical batch sizes. Only enabled when targetTable.TimestampColumn != ""; templates that don't set DedupOrderBy keep the old behaviour. Co-Authored-By: Claude Opus 4.7 --- bulker/bulkerlib/implementations/sql/snowflake.go | 15 ++++++++++++++- .../implementations/sql/sql_adapter_base.go | 1 + 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/bulker/bulkerlib/implementations/sql/snowflake.go b/bulker/bulkerlib/implementations/sql/snowflake.go index 2285b7890..1f189ac32 100644 --- a/bulker/bulkerlib/implementations/sql/snowflake.go +++ b/bulker/bulkerlib/implementations/sql/snowflake.go @@ -45,7 +45,7 @@ const ( // stages 2 and 3 read a much smaller, pre-deduped relation. Each stage // emits its own WarehouseState so timing for dedup/update/insert is // visible in the run report. - sfDedupStatement = `CREATE OR REPLACE TEMPORARY TABLE {{.NamespaceFrom}}{{.DedupTable}} AS SELECT {{.Columns}} FROM (SELECT {{.Columns}}, ROW_NUMBER() OVER (PARTITION BY {{.PrimaryKeyColumns}}{{.Discriminator}}) rn FROM {{.NamespaceFrom}}{{.TableFrom}}) QUALIFY rn = MAX(rn) OVER (PARTITION BY {{.PrimaryKeyColumns}})` + sfDedupStatement = `CREATE OR REPLACE TEMPORARY TABLE {{.NamespaceFrom}}{{.DedupTable}} AS SELECT {{.Columns}} FROM (SELECT {{.Columns}}, ROW_NUMBER() OVER (PARTITION BY {{.PrimaryKeyColumns}}{{.Discriminator}}) rn FROM {{.NamespaceFrom}}{{.TableFrom}}) QUALIFY rn = MAX(rn) OVER (PARTITION BY {{.PrimaryKeyColumns}}){{if .DedupOrderBy}} ORDER BY {{.DedupOrderBy}}{{end}}` sfMergeUpdateStatement = `UPDATE {{.Namespace}}{{.TableTo}} T SET {{.UpdateSet}} FROM {{.NamespaceFrom}}{{.DedupTable}} S WHERE {{.JoinConditions}}` sfMergeInsertStatement = `INSERT INTO {{.Namespace}}{{.TableTo}} ({{.Columns}}) SELECT {{.SourceColumns}} FROM {{.NamespaceFrom}}{{.DedupTable}} S WHERE NOT EXISTS (SELECT 1 FROM {{.Namespace}}{{.TableTo}} T WHERE {{.JoinConditions}})` sfDropDedupStatement = `DROP TABLE IF EXISTS {{.NamespaceFrom}}{{.DedupTable}}` @@ -562,12 +562,25 @@ func (s *Snowflake) copyOrMergeSplit(ctx context.Context, targetTable *Table, so }) pkColumns := utils.ArrayMap(targetTable.GetPKFields(), s.quotedColumnName) + // Pre-sort the dedup CTAS by the target's timestamp column so the + // later INSERT writes new T micro-partitions whose TO_DATE(ts) min/max + // are tight along T's CLUSTER BY (TO_DATE(timestamp)) key. This + // minimises the work auto-clustering has to do to re-partition new + // data. UPDATE rewrites preserve clustering on their own, so this is + // strictly an INSERT-side optimisation; no benefit (and no harm) + // when there is no timestamp column. + var dedupOrderBy string + if targetTable.TimestampColumn != "" { + dedupOrderBy = s.quotedColumnName(targetTable.TimestampColumn) + } + payload := QueryPayload{ Namespace: s.namespacePrefix(targetTable.Namespace), NamespaceFrom: s.namespacePrefix(sourceTable.Namespace), TableTo: s.quotedTableName(targetTable.Name), TableFrom: s.quotedTableName(sourceTable.Name), DedupTable: s.quotedTableName(sourceTable.Name + "_DEDUP"), + DedupOrderBy: dedupOrderBy, Columns: strings.Join(columnNames, ","), PrimaryKeyName: targetTable.PrimaryKeyName, PrimaryKeyColumns: strings.Join(pkColumns, ","), diff --git a/bulker/bulkerlib/implementations/sql/sql_adapter_base.go b/bulker/bulkerlib/implementations/sql/sql_adapter_base.go index 7aa574ba4..0dde8e007 100644 --- a/bulker/bulkerlib/implementations/sql/sql_adapter_base.go +++ b/bulker/bulkerlib/implementations/sql/sql_adapter_base.go @@ -428,6 +428,7 @@ type QueryPayload struct { TableTo string TableFrom string DedupTable string + DedupOrderBy string JoinConditions string SourceColumns string }