feat: ClickHouse ArchiveWriter over the fan-out scheme by Sanikadze · Pull Request #48 · open-gpdb/yagpcc

Sanikadze · 2026-07-08T20:08:42Z

Summary

Reworks the ClickHouse sink from scratch (force-pushed over the previous iteration of this PR) to plug into the ArchiveWriter interface introduced by #51 instead of carrying its own queue/batching machinery. The previous design (own ring buffer / flusher / query_events+aggregated_metrics+session_snapshots schema) is gone entirely.

What the new version does, in one line: completed queries, sessions and per-segment metrics that today go to the JSONL file archive can additionally (or instead) be streamed into ClickHouse tables yagpcc.sessions_part / yagpcc.statements_part / yagpcc.segments_part, row-for-row consistent with the JSONL contract.

Relates to / builds on #51.

What this does

Transport: implements internal/master.ArchiveWriter (StoreSessions / StoreQuery / StoreSegmensMetrics) in internal/master/clickhouse_writer.go. No sink-owned buffers, writers or backpressure - queues, deadlines and drops belong to the batch processor from Refactor sent metrics to archive - add fan out scheme #51. Each Store* call is exactly one batched native insert (PrepareBatch - Append per row - Send, clickhouse-go/v2 columnar blocks + LZ4).
DDL = production schema: writes into the tables yagpcc.sessions_part / yagpcc.statements_part / yagpcc.segments_part (columns/types/codecs/ORDER BY/PARTITION BY/TTL/skip-indexes copied verbatim from the prod ch-schema.sql). The in-repo migration is the standalone ReplacingMergeTree variant (no ON CLUSTER/Replicated/Distributed); --dump-schema --replicated prints the clustered variant. Idempotency on redelivery is guaranteed by ReplacingMergeTree + the deterministic ORDER BY.
Field mapping: source of truth is the JSONL that the Refactor sent metrics to archive - add fan out scheme #51 file writer emits (SessionDataWrite.ToJSON(), QueryStatWriteSerializable.ToJSON(), and the segment ToJSON), so ClickHouse rows are consistent with the existing Kafka/Transfer path - same JSON line, same columns. Table-driven mapping in internal/sink/clickhouse/mapping.go; unmapped JSON leaves flow into _rest. CDC columns on direct insert: _timestamp = now, _partition = 'direct', _offset = 0, _idx = row index. 64-bit identifiers (sess_id/ccnt/tm_id/query_id/plan_id) are decoded with json.Number (UseNumber) - values above 2^53 survive the JSON round-trip without float64 precision loss. template_query_text / template_plan_text are preserved (keeps the fix: propagate TemplateQueryText/TemplatePlanText through MergeQueryInfo #47 / 5d2427d fix).
Multi-target fan-out: any combination of file and/or clickhouse targets can be enabled at once. A single enabled target wires straight to the source channels; multiple targets tee each stream onto per-target channels via a non-blocking fanOut and run an independent batch-processor pipeline each (own bounded queue / write timeout / drops), so a slow or unreachable ClickHouse never stalls or drops the file writer. tm_id is stamped once in fanOut before the tee, so all targets see identical rows and writers stay read-only. writer_* pipeline metrics (writer_processed_messages_total, writer_dropped_messages_total, …) gain a target label.
Robustness: ClickHouse connects lazily - an unreachable server at startup does not crash the process; connection errors surface per batch, the pipeline drops that batch (counted in writer_dropped_messages_total{target="clickhouse"}) and the file target keeps writing losslessly. ArchiveWriter gained Close() so the ClickHouse connection is released on shutdown. Config validation rejects a clickhouse target with database != yagpcc: the DDL and inserts are fully qualified with yagpcc.*, so any other configured database would mean silently writing nowhere.

Config

writers:
  targets:
    - type: file
      enabled: true
      # ... existing file writer fields
    - type: clickhouse
      enabled: true
      addrs: ["clickhouse-01:9000", "clickhouse-02:9000"]
      database: yagpcc
      user: yagpcc
      # password via YAGPCC_CH_PASSWORD env
      tls:
        enabled: true

The YAGPCC_CH_PASSWORD env override is applied on the normal startup path as well as in the schema CLI, so the password never has to live in the YAML.

Schema management

yagpcc --dump-schema              # standalone DDL to stdout (no connection required)
yagpcc --dump-schema --replicated # clustered (ReplicatedReplacingMergeTree + ON CLUSTER + Distributed)
yagpcc --dump-migration --from=N --to=M
yagpcc --migrate-only             # apply standalone migrations and exit
yagpcc --verify-schema            # check schema version and exit

--migrate-only connects through database=default rather than the configured database: on a fresh server the yagpcc database does not exist yet, and pinning the connection to it made bootstrap fail with code: 81 before CREATE DATABASE could run. All DDL statements are fully qualified (yagpcc.*), so the session database does not matter; regular runtime inserts still connect to the configured database, and the database == yagpcc validation guard is unaffected.

Docs: docs/historical-stats-flow.md gains a "ClickHouse target" section.

Open questions

CDC column values on direct insert (_partition='direct', _offset=0, _idx=row index, _rest=remainder) - happy to adjust to whatever is preferred.
Per-target write timeouts / queue sizes are currently shared with the file target's settings; open to a per-target override knob if wanted.

Task 1 of the ClickHouse ArchiveWriter plan: bring over only the reusable pieces from feat/clickhouse-sink (516ba69) so they compile against main+open-gpdb#51. Transferred: connection wrapper (client.go), migration machinery (migrations.go/schema.go) and its schema CLI (schema_cli.go), and the ClickHouse config (config/clickhouse.go), all with their tests. metrics.go is trimmed to the insert-oriented collectors (inserts_total, batch_duration_seconds, dropped_rows_total with mapping_error/insert_error reasons) — the buffer/writer-coupled hooks and gauges are dropped. Not transferred: buffer.go/writer.go/tables.go/mapping.go and the old background.go wiring. The integration test is deferred to Task 6 (needs testcontainers-go and the new DDL). The old query_events DDL rides along as the migration substrate and is rewritten to the production schema in Task 2. Wired the schema CLI into cmd/server/main.go and added a top-level clickhouse config block (validated for the master role). make build and make unittest are green.

Rewrote 0001_init.up/down.sql as the standalone variant of the production schema: tables yagpcc.sessions_part/statements_part/segments_part with columns, types, codecs, ORDER BY/PARTITION BY/TTL taken verbatim from ch-schema.sql, engine ReplacingMergeTree without ON CLUSTER/Distributed. Skipping indexes (minmax) are built right into CREATE TABLE. Migrations are now text/template: {{if .Replicated}} switches between the standalone and clustered (ReplicatedReplacingMergeTree + ON CLUSTER + Distributed) variants. schema_cli gained a --replicated flag to print the clustered variant; ApplyMigrations/--migrate-only always renders standalone. TTLs are fixed (60/180 days as in production) — RetentionDays is no longer substituted into the DDL. Tests updated for the new schema, plus checks for both variants and for matching column sets between standalone and replicated.

Rewrite internal/sink/clickhouse/mapping.go as a table-driven mapper from the file-writer JSON streams onto the production schema tables sessions_part / statements_part / segments_part. Path elements are alias-sets so the same GPMetrics leaf resolves under both encoding/json (sessions, snake_case) and protojson (queries/segments, camelCase), and coercion accepts protojson's string-encoded 64-bit ints. Unmapped fields flow into _rest; CDC columns are filled per the plan (_partition="direct", _offset=0, _idx=row index).

Add internal/master/clickhouse_writer.go implementing ArchiveWriter with strictly batched native inserts (PrepareBatch/Append/Send) for the sessions/statements/segments streams. Mapping failures are skipped as mapping_error; PrepareBatch/Append/Send failures are wrapped and returned so the batch pipeline drops the batch. Export clickhouse.Mapping alias and Table() so master can hold the per-table mappings.

Add clickhouse target type to writers.targets (addrs/database/TLS, password from env) with validation. Implement independent per-target batch-processor pipelines: a single enabled target wires straight to the source channels; multiple targets tee each stream onto per-target channels via a non-blocking fanOut so a slow/unreachable target only drops its own messages. Add a target label to the writer_* pipeline metrics.

…-out race Task 6: add a real-ClickHouse integration test (build tag `integration`, gated on YAGPCC_CH_ADDR) that applies the standalone production DDL and drives the sessions/statements/segments mapping through the same PrepareBatch/Send path as the master writer, verifying key columns, CDC meta, _rest routing, preserved template texts, and ReplacingMergeTree idempotency (OPTIMIZE/SELECT FINAL). Wire a clickhouse-integration CI job backed by a ClickHouse service container. Document the clickhouse writer target in historical-stats-flow.md. Also fix a data race introduced by the Task 5 multi-target fan-out: the same message pointer was teed to several pipelines while both the file and ClickHouse writers mutated TmID concurrently. Stamp the discovered TmID once in fanOut (the sole owner before the tee) and make the writers read-only; a lone target now flows through a blocking fanOut so source backpressure is preserved.

leborchuk · 2026-07-15T08:19:26Z

Need to be rewritten as was suggested in f74a5f5

- fanout: read gp.DiscoveredTmID via atomic.LoadInt64 to match the atomic CAS writer and stay race-free under --race - config: apply YAGPCC_CH_PASSWORD env override in ReadFromFile before Validate so a master with the ClickHouse secret only in the env var no longer fails startup with "password is required" - README: document the ClickHouse archive sink, schema-management CLI flags, YAGPCC_CH_PASSWORD, and link historical-stats-flow.md

The session stream is serialised with encoding/json, which emits uint64 query_id/plan_id as bare JSON numbers. MapRow decoded into map[string]any without UseNumber, so those ids were parsed as float64 and rounded above 2^53 before reaching ClickHouse. Decode with UseNumber and parse the resulting json.Number losslessly in the numeric coercion helpers.

- Reject non-"yagpcc" ClickHouse database in config validation. The writer qualifies INSERTs with the configured database, but the embedded DDL/migrations only create tables in "yagpcc"; a non-default value would insert into an uncreated database and drop every batch silently. Guard both the top-level clickhouse block and writer targets. - Add Close() to ArchiveWriter so master restarts / leadership changes no longer orphan ClickHouse connection pools or file descriptors. FileWriters closes its rotating files (RotateWriter gains Close), ClickHouseWriters closes the connection, and launchArchiveWriters closes all writers on context cancellation. buildArchiveWriters now closes already-built writers when a later target fails. - Correct the multi-target fan-out comment: with >1 target sends are non-blocking and drop on a full channel (they do not back up).

Without CLICKHOUSE_USER/CLICKHOUSE_PASSWORD the clickhouse-server image entrypoint restricts the default user to localhost inside the container, so the integration job fails with AUTHENTICATION_FAILED when connecting through the mapped port. CLICKHOUSE_SKIP_USER_SETUP=1 keeps the stock users.xml with unrestricted network access for the passwordless default user.

On a fresh ClickHouse server the yagpcc database does not exist until RunMigrations executes its CREATE DATABASE IF NOT EXISTS bootstrap, but the connection DSN was pinned to the configured (still missing) target database, so the ping failed with UNKNOWN_DATABASE before any SQL could run and --migrate-only could never bootstrap from scratch. Open the schema CLI connection through the server default database instead: every schema statement is fully qualified (yagpcc.*), so the session database is irrelevant. Validate still runs against the configured database first, keeping the database==yagpcc guard intact. Found on a stand deploy: DROP DATABASE yagpcc + --migrate-only failed with 'code: 81 ... Database yagpcc does not exist' (the integration test did not catch this because it connects through default already).

Sanikadze · 2026-08-01T19:41:27Z

@leborchuk Reworked per #51 (ArchiveWriter + ch-schema.sql DDL) and force-pushed, description updated; CI green, ready for review.

Copilot

Pull request overview

Adds a new ClickHouse archive target to yagpcc’s master-side historical telemetry pipeline, implementing the ArchiveWriter interface so the existing batch processor (from #51) can fan out the same session/statement/segment JSONL contract to ClickHouse tables and to the existing file archive.

Changes:

Implement ClickHouse writer (internal/master/clickhouse_writer.go) + fan-out plumbing (internal/master/fanout.go, background.go) to support independent per-target pipelines.
Introduce ClickHouse sink package (internal/sink/clickhouse/) with native inserts, JSON→column mapping, metrics, embedded schema migrations, and an integration test (tagged).
Add schema-management CLI (--dump-schema, --migrate-only, --verify-schema, etc.), config surface area, docs updates, and CI job for ClickHouse integration tests.

Reviewed changes

Copilot reviewed 35 out of 38 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
README.md	Documents ClickHouse archiving + schema-management CLI flags.
internal/sink/clickhouse/testdata/.gitkeep	Keeps testdata dir in VCS.
internal/sink/clickhouse/schema.go	Adds schema verification and schema/migration dump helpers over embedded migrations.
internal/sink/clickhouse/schema_test.go	Unit tests for schema verification and dump rendering (standalone vs replicated).
internal/sink/clickhouse/migrations/0001_init.up.sql	Adds initial ClickHouse DDL for `yagpcc.*_part` tables + replicated variants.
internal/sink/clickhouse/migrations/0001_init.down.sql	Adds down migration for dropping ClickHouse schema objects.
internal/sink/clickhouse/migrations/.gitkeep	Keeps migrations dir in VCS.
internal/sink/clickhouse/migrations.go	Implements embedded migration parsing, templating, splitting, and apply logic.
internal/sink/clickhouse/migrations_test.go	Unit tests for migration parsing/templating/splitting.
internal/sink/clickhouse/migration_runner_test.go	Unit tests for GetCurrentVersion / ApplyMigrations behavior using a fake conn.
internal/sink/clickhouse/metrics.go	Adds Prometheus metrics for ClickHouse batch inserts and dropped rows.
internal/sink/clickhouse/metrics_test.go	Unit tests for ClickHouse metrics registration and updates.
internal/sink/clickhouse/mapping.go	Implements table-driven JSON→ClickHouse column mapping with `_rest` capture.
internal/sink/clickhouse/mapping_test.go	Golden/unit tests for mapping correctness and precision preservation.
internal/sink/clickhouse/integration_test.go	Integration test (build tag) exercising DDL + native inserts against real ClickHouse.
internal/sink/clickhouse/client.go	ClickHouse connection wrapper (TLS, settings, ping).
internal/sink/clickhouse/client_test.go	Unit tests for TLS config validation and client/ping behavior.
internal/master/file_writer.go	Adds `Close()` to file writers/rotate writer; removes tm_id stamping from writer.
internal/master/fanout.go	Adds generic fan-out with per-target drop accounting + tm_id stamping helpers.
internal/master/fanout_test.go	Tests fan-out delivery and multi-target “slow target doesn’t block” behavior.
internal/master/clickhouse_writer.go	New ArchiveWriter implementation that maps JSONL → ClickHouse native batches.
internal/master/clickhouse_writer_test.go	Unit tests for ClickHouse writer behavior and error/metrics paths using fakes.
internal/master/batch_processor.go	Adds `target` label plumbed through batch processors and writer pipeline metrics.
internal/master/batch_processor_test.go	Updates tests to pass the new `target` parameter.
internal/master/background.go	Wires multi-target archive writers, per-target batch processors, and fan-out stamping.
internal/master/archive_writer.go	Extends `ArchiveWriter` with `Close()` for resource cleanup.
internal/config/config.go	Adds ClickHouse config block + clickhouse target fields, env password override, validation.
internal/config/config_test.go	Adds tests for clickhouse target validation and target→ClickhouseConfig mapping.
internal/config/clickhouse.go	Introduces `ClickhouseConfig` (+ defaults, env overrides, validation) and constants.
internal/config/clickhouse_test.go	Adds comprehensive tests for ClickhouseConfig defaults/validation/env override.
internal/app/app.go	Extends writer pipeline metrics with a `target` label.
go.mod	Adds ClickHouse driver dependency and required indirect deps.
docs/historical-stats-flow.md	Documents writer targets fan-out and ClickHouse target configuration.
cmd/server/schema_cli.go	Adds schema-management CLI implementation (dump/migrate/verify, replicated option).
cmd/server/schema_cli_test.go	Unit tests for schema CLI flag dispatch and behavior.
cmd/server/main.go	Routes schema CLI commands before starting the main app loop.
.github/workflows/test.yaml	Adds ClickHouse integration test job using a service container.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

+func uint64FromString(v string) (uint64, error) {
+	if v == "" {
+		return 0, nil
+	}
+	if n, err := strconv.ParseUint(v, 10, 64); err == nil {
+		return n, nil
+	}
+	f, err := strconv.ParseFloat(v, 64)
+	if err != nil {
+		return 0, fmt.Errorf("cannot convert %q to uint64", v)
+	}
+	return uint64(f), nil
+}


+func int64FromString(v string) (int64, error) {
+	if v == "" {
+		return 0, nil
+	}
+	if n, err := strconv.ParseInt(v, 10, 64); err == nil {
+		return n, nil
+	}
+	f, err := strconv.ParseFloat(v, 64)
+	if err != nil {
+		return 0, fmt.Errorf("cannot convert %q to int64", v)
+	}
+	return int64(f), nil
+}


+	case kindU64:
+		return toUint64(raw)
+	case kindU32:
+		n, err := toUint64(raw)
+		return uint32(n), err
+	case kindI64:


+
+	ts := time.Now()
+	rows := make([][]any, 0, len(jsons))
+	for i, js := range jsons {
+		meta := clickhouse.CDCMeta{
+			Timestamp: ts,


leborchuk

Manually checked it on my environment - works good

leborchuk · 2026-08-12T14:49:45Z

Need to fix issues found by Copilot

Sanikadze · 2026-08-15T19:25:02Z

@leborchuk Fixed the Copilot findings and force-pushed: strict integer parsing, range checks for u32/i32/enum coercions, _timestamp in UTC, negative values in nullable unsigned fields mapped to NULL (Unix-socket sessions have ClientPort=-1). The down-migration finding was a false positive-the guard was already correct.

Sanikadze added 2 commits July 7, 2026 20:14

Sanikadze force-pushed the feat/clickhouse-sink branch from b3ed724 to ac9f378 Compare July 9, 2026 21:21

Sanikadze added 4 commits July 10, 2026 19:57

Sanikadze added 5 commits July 15, 2026 22:56

Sanikadze force-pushed the feat/clickhouse-sink branch 2 times, most recently from 00f70a2 to c6d91db Compare August 1, 2026 19:25

leborchuk requested a lite review from Copilot August 12, 2026 08:06

Copilot started reviewing on behalf of leborchuk August 12, 2026 08:11 View session

Copilot AI reviewed Aug 12, 2026

View reviewed changes

leborchuk approved these changes Aug 12, 2026

View reviewed changes

Sanikadze added 2 commits August 15, 2026 18:11

style: reformat replicated guards in down migration

a0b1a2c

fix: address Copilot review findings

ed7213f

Sanikadze changed the title ~~feat: ClickHouse sink for query history, aggregated metrics and session snapshots~~ feat: ClickHouse ArchiveWriter over the fan-out scheme Aug 15, 2026

Sanikadze force-pushed the feat/clickhouse-sink branch from 994c35a to ed7213f Compare August 15, 2026 19:22

leborchuk merged commit 8a9a45a into open-gpdb:main Aug 17, 2026
3 checks passed

Navigation Menu

Uh oh!

feat: ClickHouse ArchiveWriter over the fan-out scheme - #48

feat: ClickHouse ArchiveWriter over the fan-out scheme#48
leborchuk merged 13 commits into
open-gpdb:mainfrom
Sanikadze:feat/clickhouse-sink

Sanikadze commented Jul 8, 2026 •

edited

Loading

Uh oh!

leborchuk commented Jul 15, 2026

Uh oh!

Sanikadze commented Aug 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

leborchuk left a comment

Uh oh!

leborchuk commented Aug 12, 2026

Uh oh!

Sanikadze commented Aug 15, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

Sanikadze commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

What this does

Config

Schema management

Open questions

Uh oh!

leborchuk commented Jul 15, 2026

Uh oh!

Sanikadze commented Aug 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

leborchuk left a comment

Choose a reason for hiding this comment

Uh oh!

leborchuk commented Aug 12, 2026

Uh oh!

Sanikadze commented Aug 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Sanikadze commented Jul 8, 2026 •

edited

Loading

Sanikadze commented Aug 15, 2026 •

edited

Loading