Skip to content

Commit 4e95b70

Browse files
committed
Updated docs and readme
1 parent 006ed2e commit 4e95b70

16 files changed

Lines changed: 704 additions & 371 deletions

File tree

AGENTS.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# AGENTS.md
2+
3+
## Verification
4+
5+
Before finishing any code change, always run:
6+
7+
```sh
8+
go mod tidy
9+
go test ./...
10+
./e2e/run.sh
11+
```
12+
13+
If any command cannot be run, report that explicitly with the reason.

README.md

Lines changed: 39 additions & 282 deletions
Large diffs are not rendered by default.

cmd/partforge/main.go

Lines changed: 77 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -842,6 +842,10 @@ func runWorker(ctx context.Context, args []string) error {
842842
if err != nil {
843843
return fmt.Errorf("derive clickhouse merge background pool size: %w", err)
844844
}
845+
mergeConcurrencyRatio, maxConcurrentMerges, err := resources.MergeConcurrencyRatio(workerLimits, mergeBackgroundPoolSize)
846+
if err != nil {
847+
return fmt.Errorf("derive clickhouse merge concurrency ratio: %w", err)
848+
}
845849
sourceMergeIdleTimeout, sourceMergeMaxRuntime := sourceMergeWaitTimeouts(*mergeIdleTimeout, *mergeMaxRuntime, roleSettings.SourceMergeCompactCap)
846850
sourceMergeSettleMinWait := derivedMergeSettleMinWait(sourceMergeIdleTimeout, rewrite.DefaultMergeSettleMinWait)
847851
compactStaleAfter := compactLeaseStaleAfter(*compactWindow)
@@ -860,9 +864,12 @@ func runWorker(ctx context.Context, args []string) error {
860864
"min_insert_block_size_bytes_raw", insertSettings["min_insert_block_size_bytes"],
861865
"default_compression_codec", *defaultCompressionCodec,
862866
"merge_background_pool_size", mergeBackgroundPoolSize,
867+
"merge_concurrency_ratio", mergeConcurrencyRatio,
868+
"merge_max_concurrent_merges", maxConcurrentMerges,
863869
"merge_max_block_size", mergeTreeSettings.MergeMaxBlockSize,
864870
"merge_max_block_size_bytes", mergeTreeSettings.MergeMaxBlockSizeBytes,
865871
"merge_selecting_sleep_ms", mergeTreeSettings.MergeSelectingSleepMS,
872+
"merge_pool_free_entries_threshold", mergeTreeSettings.PoolFreeEntriesThreshold,
866873
"background_merges_mutations_scheduling_policy", mergeTreeSettings.MergeSchedulingPolicy,
867874
"merge_idle_timeout", sourceMergeIdleTimeout,
868875
"merge_max_runtime", sourceMergeMaxRuntime,
@@ -921,32 +928,34 @@ func runWorker(ctx context.Context, args []string) error {
921928
if part == nil {
922929
if roleSettings.Compact {
923930
didCompactWork, err := runWorkerCompaction(ctx, workerCompactionConfig{
924-
StateStore: stateStore,
925-
WorkerID: resolvedWorkerID,
926-
WorkDir: *workDir,
927-
ClickHouseURL: *clickHouseURL,
928-
ClickHouseUser: *clickHouseUser,
929-
ClickHousePassword: *clickHousePassword,
930-
ClickHouseBinary: *clickHouseBinary,
931-
ClickHouseConfigFile: *clickHouseConfigFile,
932-
ClickHousePrometheus: clickHousePrometheusConfig,
933-
ClickHousePrometheusTarget: clickHousePrometheusTarget,
934-
S5cmdBinary: *s5cmdBinary,
935-
S3Endpoint: *s3Endpoint,
936-
DefaultCompressionCodec: *defaultCompressionCodec,
937-
MergeBackgroundPoolSize: mergeBackgroundPoolSize,
938-
MergeSchedulingPolicy: mergeTreeSettings.MergeSchedulingPolicy,
939-
MergeMaxBlockSize: mergeTreeSettings.MergeMaxBlockSize,
940-
MergeMaxBlockSizeBytes: mergeTreeSettings.MergeMaxBlockSizeBytes,
941-
MergeSelectingSleepMS: mergeTreeSettings.MergeSelectingSleepMS,
942-
CompactWindow: *compactWindow,
943-
CompactOptimizeFinalAfter: effectiveCompactOptimizeFinalAfter,
944-
CompactLeaseStaleAfter: compactStaleAfter,
945-
CompactHeartbeatInterval: compactHeartbeatInterval,
946-
CompactMaxArtifacts: *compactMaxArtifacts,
947-
CompactMaxBytes: *compactMaxBytes,
948-
Metrics: recorder,
949-
PrometheusMetrics: prometheusMetrics,
931+
StateStore: stateStore,
932+
WorkerID: resolvedWorkerID,
933+
WorkDir: *workDir,
934+
ClickHouseURL: *clickHouseURL,
935+
ClickHouseUser: *clickHouseUser,
936+
ClickHousePassword: *clickHousePassword,
937+
ClickHouseBinary: *clickHouseBinary,
938+
ClickHouseConfigFile: *clickHouseConfigFile,
939+
ClickHousePrometheus: clickHousePrometheusConfig,
940+
ClickHousePrometheusTarget: clickHousePrometheusTarget,
941+
S5cmdBinary: *s5cmdBinary,
942+
S3Endpoint: *s3Endpoint,
943+
DefaultCompressionCodec: *defaultCompressionCodec,
944+
MergeBackgroundPoolSize: mergeBackgroundPoolSize,
945+
MergeConcurrencyRatio: mergeConcurrencyRatio,
946+
MergeSchedulingPolicy: mergeTreeSettings.MergeSchedulingPolicy,
947+
MergeMaxBlockSize: mergeTreeSettings.MergeMaxBlockSize,
948+
MergeMaxBlockSizeBytes: mergeTreeSettings.MergeMaxBlockSizeBytes,
949+
MergeSelectingSleepMS: mergeTreeSettings.MergeSelectingSleepMS,
950+
MergePoolFreeEntriesThreshold: mergeTreeSettings.PoolFreeEntriesThreshold,
951+
CompactWindow: *compactWindow,
952+
CompactOptimizeFinalAfter: effectiveCompactOptimizeFinalAfter,
953+
CompactLeaseStaleAfter: compactStaleAfter,
954+
CompactHeartbeatInterval: compactHeartbeatInterval,
955+
CompactMaxArtifacts: *compactMaxArtifacts,
956+
CompactMaxBytes: *compactMaxBytes,
957+
Metrics: recorder,
958+
PrometheusMetrics: prometheusMetrics,
950959
})
951960
if err != nil {
952961
return err
@@ -1064,10 +1073,11 @@ func runWorker(ctx context.Context, args []string) error {
10641073
InsertSettings: insertSettings,
10651074
ProgressInterval: *stateProgressInterval,
10661075
MergeTreeSettings: rewrite.MergeTreeSettings{
1067-
MergeMaxBlockSize: mergeTreeSettings.MergeMaxBlockSize,
1068-
MergeMaxBlockSizeBytes: mergeTreeSettings.MergeMaxBlockSizeBytes,
1069-
MergeSelectingSleepMS: mergeTreeSettings.MergeSelectingSleepMS,
1070-
DefaultCompressionCodec: *defaultCompressionCodec,
1076+
MergeMaxBlockSize: mergeTreeSettings.MergeMaxBlockSize,
1077+
MergeMaxBlockSizeBytes: mergeTreeSettings.MergeMaxBlockSizeBytes,
1078+
MergeSelectingSleepMS: mergeTreeSettings.MergeSelectingSleepMS,
1079+
DefaultCompressionCodec: *defaultCompressionCodec,
1080+
PoolFreeEntriesThreshold: mergeTreeSettings.PoolFreeEntriesThreshold,
10711081
},
10721082
}
10731083
processor.RestartClickHouse = func(ctx context.Context) error {
@@ -1080,8 +1090,8 @@ func runWorker(ctx context.Context, args []string) error {
10801090
return fmt.Errorf("stop clickhouse before restart: %w", err)
10811091
}
10821092
server = nil
1083-
slog.Info("starting local ClickHouse server after restart", "stage", "restart_clickhouse", "binary", *clickHouseBinary, "config_file", *clickHouseConfigFile, "clickhouse_data_dir", runDirs.ClickHouse, "job_id", part.JobID, "part_id", part.PartID, "background_pool_size", mergeBackgroundPoolSize, "background_merges_mutations_scheduling_policy", mergeTreeSettings.MergeSchedulingPolicy)
1084-
restarted, err := startServer(ctx, chproc.Tuning{BackgroundPoolSize: mergeBackgroundPoolSize, MergeSchedulingPolicy: mergeTreeSettings.MergeSchedulingPolicy})
1093+
slog.Info("starting local ClickHouse server after restart", "stage", "restart_clickhouse", "binary", *clickHouseBinary, "config_file", *clickHouseConfigFile, "clickhouse_data_dir", runDirs.ClickHouse, "job_id", part.JobID, "part_id", part.PartID, "background_pool_size", mergeBackgroundPoolSize, "background_merges_mutations_concurrency_ratio", mergeConcurrencyRatio, "background_merges_mutations_scheduling_policy", mergeTreeSettings.MergeSchedulingPolicy)
1094+
restarted, err := startServer(ctx, chproc.Tuning{BackgroundPoolSize: mergeBackgroundPoolSize, MergeConcurrencyRatio: mergeConcurrencyRatio, MergeSchedulingPolicy: mergeTreeSettings.MergeSchedulingPolicy})
10851095
if err != nil {
10861096
return err
10871097
}
@@ -1195,32 +1205,34 @@ func createWorkerRunDirs(workDir string) (workerRunDirs, error) {
11951205
}
11961206

11971207
type workerCompactionConfig struct {
1198-
StateStore *state.Store
1199-
WorkerID string
1200-
WorkDir string
1201-
ClickHouseURL string
1202-
ClickHouseUser string
1203-
ClickHousePassword string
1204-
ClickHouseBinary string
1205-
ClickHouseConfigFile string
1206-
ClickHousePrometheus chproc.PrometheusConfig
1207-
ClickHousePrometheusTarget string
1208-
S5cmdBinary string
1209-
S3Endpoint string
1210-
DefaultCompressionCodec string
1211-
MergeBackgroundPoolSize int
1212-
MergeSchedulingPolicy string
1213-
MergeMaxBlockSize uint64
1214-
MergeMaxBlockSizeBytes uint64
1215-
MergeSelectingSleepMS uint64
1216-
CompactWindow time.Duration
1217-
CompactOptimizeFinalAfter time.Duration
1218-
CompactLeaseStaleAfter time.Duration
1219-
CompactHeartbeatInterval time.Duration
1220-
CompactMaxArtifacts int
1221-
CompactMaxBytes uint64
1222-
Metrics metrics.Recorder
1223-
PrometheusMetrics *metrics.Prometheus
1208+
StateStore *state.Store
1209+
WorkerID string
1210+
WorkDir string
1211+
ClickHouseURL string
1212+
ClickHouseUser string
1213+
ClickHousePassword string
1214+
ClickHouseBinary string
1215+
ClickHouseConfigFile string
1216+
ClickHousePrometheus chproc.PrometheusConfig
1217+
ClickHousePrometheusTarget string
1218+
S5cmdBinary string
1219+
S3Endpoint string
1220+
DefaultCompressionCodec string
1221+
MergeBackgroundPoolSize int
1222+
MergeConcurrencyRatio float64
1223+
MergeSchedulingPolicy string
1224+
MergeMaxBlockSize uint64
1225+
MergeMaxBlockSizeBytes uint64
1226+
MergeSelectingSleepMS uint64
1227+
MergePoolFreeEntriesThreshold uint64
1228+
CompactWindow time.Duration
1229+
CompactOptimizeFinalAfter time.Duration
1230+
CompactLeaseStaleAfter time.Duration
1231+
CompactHeartbeatInterval time.Duration
1232+
CompactMaxArtifacts int
1233+
CompactMaxBytes uint64
1234+
Metrics metrics.Recorder
1235+
PrometheusMetrics *metrics.Prometheus
12241236
}
12251237

12261238
func runWorkerCompaction(ctx context.Context, cfg workerCompactionConfig) (bool, error) {
@@ -1533,10 +1545,11 @@ func processCompactBatch(ctx, shutdownCtx, manualFinalizeCtx context.Context, cf
15331545
MergeDeadline: compactDeadline,
15341546
OptimizeFinalAfter: cfg.CompactOptimizeFinalAfter,
15351547
MergeTreeSettings: rewrite.MergeTreeSettings{
1536-
MergeMaxBlockSize: cfg.MergeMaxBlockSize,
1537-
MergeMaxBlockSizeBytes: cfg.MergeMaxBlockSizeBytes,
1538-
MergeSelectingSleepMS: cfg.MergeSelectingSleepMS,
1539-
DefaultCompressionCodec: cfg.DefaultCompressionCodec,
1548+
MergeMaxBlockSize: cfg.MergeMaxBlockSize,
1549+
MergeMaxBlockSizeBytes: cfg.MergeMaxBlockSizeBytes,
1550+
MergeSelectingSleepMS: cfg.MergeSelectingSleepMS,
1551+
DefaultCompressionCodec: cfg.DefaultCompressionCodec,
1552+
PoolFreeEntriesThreshold: cfg.MergePoolFreeEntriesThreshold,
15401553
},
15411554
ShutdownContext: shutdownCtx,
15421555
MergeStopContext: manualFinalizeCtx,
@@ -1566,8 +1579,8 @@ func processCompactBatch(ctx, shutdownCtx, manualFinalizeCtx context.Context, cf
15661579
return fmt.Errorf("stop clickhouse before compact restart: %w", err)
15671580
}
15681581
server = nil
1569-
slog.Info("starting local ClickHouse server after compact restart", "stage", "compact_restart_clickhouse", "binary", cfg.ClickHouseBinary, "config_file", cfg.ClickHouseConfigFile, "clickhouse_data_dir", runDirs.ClickHouse, "job_id", item.JobID, "output_part_id", item.OutputPartID, "background_pool_size", cfg.MergeBackgroundPoolSize, "background_merges_mutations_scheduling_policy", cfg.MergeSchedulingPolicy)
1570-
restarted, err := startServer(ctx, chproc.Tuning{BackgroundPoolSize: cfg.MergeBackgroundPoolSize, MergeSchedulingPolicy: cfg.MergeSchedulingPolicy})
1582+
slog.Info("starting local ClickHouse server after compact restart", "stage", "compact_restart_clickhouse", "binary", cfg.ClickHouseBinary, "config_file", cfg.ClickHouseConfigFile, "clickhouse_data_dir", runDirs.ClickHouse, "job_id", item.JobID, "output_part_id", item.OutputPartID, "background_pool_size", cfg.MergeBackgroundPoolSize, "background_merges_mutations_concurrency_ratio", cfg.MergeConcurrencyRatio, "background_merges_mutations_scheduling_policy", cfg.MergeSchedulingPolicy)
1583+
restarted, err := startServer(ctx, chproc.Tuning{BackgroundPoolSize: cfg.MergeBackgroundPoolSize, MergeConcurrencyRatio: cfg.MergeConcurrencyRatio, MergeSchedulingPolicy: cfg.MergeSchedulingPolicy})
15711584
if err != nil {
15721585
return err
15731586
}

docs/deployment.md

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
# Deployment
2+
3+
The worker image is published on every push to `main` to `ghcr.io/<owner>/partforge`, tagged with the short commit SHA and `latest`. CI also attaches static `linux/amd64` and `linux/arm64` CLI binaries to a GitHub release named after the SHA.
4+
5+
The image is a single Ubuntu container with `clickhouse-server`, `clickhouse-client`, `s5cmd`, and the Go binary. Its entrypoint is the binary and the default command is `worker`. It **runs as root** (so it can write its work directory on root-owned host mounts) and starts a local `clickhouse server` child process for each claimed part.
6+
7+
## Recommended: workers on ECS with an IAM task role
8+
9+
Run the workers as an ECS service and give the task an **IAM role** scoped to the S3 bucket and the DynamoDB table. This is the recommended setup:
10+
11+
- **No static credentials.** The AWS SDK picks up temporary credentials from the ECS task role via the container credentials endpoint. Do not bake access keys into the image or config.
12+
- **Region resolves from the environment.** Set `AWS_REGION` (or `-aws-region`); otherwise it falls back through AWS config, IMDS, then `us-east-1`.
13+
- **Scale by replicas.** More worker tasks = more parts in flight. There is no coordinator to scale — workers claim independently from DynamoDB.
14+
15+
### Task IAM policy
16+
17+
Combine the S3 and DynamoDB permissions. DynamoDB detail (including tighter per-role variants) is in [dynamodb.md](dynamodb.md).
18+
19+
```json
20+
{
21+
"Version": "2012-10-17",
22+
"Statement": [
23+
{
24+
"Effect": "Allow",
25+
"Action": [
26+
"dynamodb:Query",
27+
"dynamodb:Scan",
28+
"dynamodb:PutItem",
29+
"dynamodb:UpdateItem",
30+
"dynamodb:DeleteItem",
31+
"dynamodb:TransactWriteItems"
32+
],
33+
"Resource": [
34+
"arn:aws:dynamodb:us-east-1:123456789012:table/partforge",
35+
"arn:aws:dynamodb:us-east-1:123456789012:table/partforge/index/*"
36+
]
37+
},
38+
{
39+
"Effect": "Allow",
40+
"Action": ["s3:ListBucket"],
41+
"Resource": "arn:aws:s3:::partforge"
42+
},
43+
{
44+
"Effect": "Allow",
45+
"Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
46+
"Resource": "arn:aws:s3:::partforge/*"
47+
}
48+
]
49+
}
50+
```
51+
52+
`s3:DeleteObject` is needed because workers replace finished-artifact prefixes (`s5cmd rm` then upload) and admin `-delete-s3` operations remove artifacts.
53+
54+
### Storage matters
55+
56+
Worker scratch (`-work-dir`) holds the local ClickHouse data plus downloaded source parts, and compaction transiently holds downloaded tarballs, extracted parts, merge output, and re-uploaded tarballs at once. It must be **fast local disk with enough headroom**:
57+
58+
- **EC2 launch type with instance-store NVMe** is best for large parts — mount the NVMe into the container and set `-work-dir` on it (e.g. `/mnt/nvme/partforge-work`).
59+
- **Fargate** works for smaller jobs; raise the task's ephemeral storage and keep `-compact-max-bytes` well below it.
60+
61+
Each claimed part gets its own `run-*` directory that is removed when the part finishes.
62+
63+
### Splitting inserter and compactor
64+
65+
Run the rewrite and compaction stages as separate services to scale them independently:
66+
67+
- `worker -role=inserter` — rewrite only.
68+
- `worker -role=compactor` — compaction only.
69+
- `worker -role=all` (default) — rewrite first, compact when idle.
70+
71+
See [operations.md](operations.md) for the full flag set and metrics.
72+
73+
## Where the other commands run
74+
75+
`worker` is the only stage that belongs on ECS. The other two need local access to a ClickHouse node's disks and generally run there:
76+
77+
- **`upload-freeze`** must run where it can read the source ClickHouse data disks reported by `system.disks`.
78+
- **`import-finished`** must run where its work-dir shares a filesystem with the destination table's `detached` directory (parts are moved, not copied).
79+
80+
Both still need the same S3 + DynamoDB access as the workers.

docs/development.md

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
# Development
2+
3+
## Checks
4+
5+
CI (`.github/workflows/release.yml`) enforces formatting, tidy modules, vet, and tests. Run what CI runs before finishing a change:
6+
7+
```sh
8+
gofmt -l . # must print nothing
9+
go mod tidy # must not change go.mod / go.sum
10+
go vet ./...
11+
go test ./...
12+
./e2e/run.sh # requires Docker
13+
```
14+
15+
`AGENTS.md` lists `go mod tidy && go test ./... && ./e2e/run.sh` as the required pre-commit check.
16+
17+
The e2e script stands up LocalStack + a ClickHouse container, builds the worker image, and runs the full pipeline against `e2e/sql/`, diffing the result against `e2e/expected.tsv`. It builds the image each run; set `PARTFORGE_E2E_SKIP_BUILD=1` to reuse an existing `partforge-worker:latest`.
18+
19+
## Build
20+
21+
```sh
22+
go build -o partforge ./cmd/partforge # local CLI
23+
docker compose build worker # worker image
24+
```
25+
26+
## Project layout
27+
28+
```
29+
cmd/partforge/ CLI entrypoint — every subcommand, flag parsing, config resolution
30+
internal/
31+
freeze/ discover ClickHouse disks; scan shadow/<freeze> for frozen parts
32+
manifest/ per-part manifest.json; job/part ID derivation
33+
artifact/ write manifests; build/extract part tarballs
34+
s3copy/ s5cmd wrapper for directory/glob transfers
35+
state/ DynamoDB state store — claims, transitions, compaction batches, admin ops
36+
chproc/ start/stop the local clickhouse-server child process
37+
chhttp/ ClickHouse HTTP client
38+
ddl/ CREATE TABLE normalization (Replicated* -> plain MergeTree)
39+
rewrite/ the worker: per-part processor + compactor
40+
resources/ CPU/memory detection -> ClickHouse insert & merge tuning
41+
parts/ import-finished: attach part tarballs into the destination table
42+
metrics/ Prometheus recorder + metrics HTTP server
43+
fileutil/ filesystem copy and directory stats
44+
```
45+
46+
Where to change things:
47+
48+
- rewrite / merge-wait / compaction logic → `internal/rewrite`
49+
- state transitions and admin operations → `internal/state`
50+
- ClickHouse tuning heuristics → `internal/resources`
51+
- CLI flags and command wiring → `cmd/partforge/main.go`
52+
53+
## Release
54+
55+
On push to `main`, CI builds and publishes the multi-arch worker image to `ghcr.io/<owner>/partforge` (tagged with the short commit SHA and `latest`) and attaches static `linux/amd64` and `linux/arm64` CLI binaries to a GitHub release named after the SHA. See [deployment.md](deployment.md).

0 commit comments

Comments
 (0)