diff --git a/README.md b/README.md index 49517b3..defe7a3 100644 --- a/README.md +++ b/README.md @@ -646,7 +646,14 @@ Assignment-driven environment groups add a middle layer for objects shared by an exact subset of environments. A group declaration is written once under `layers/group//` and inserted between the shared and environment layers in every member's manifest stack. Group members must contain identical -copies of the object; presence or content drift fails closed. +copies of non-table objects. Tables may differ in settings only; other content +or presence drift fails closed. + +For tables in shared or group placement, decompose synthesizes the base +settings as the key-and-value intersection across the placement's member +environments. Settings outside that intersection become additive environment +patches, including keys present everywhere with different values. This avoids +subtractive settings patches while preserving each environment exactly. Engine block differences that `patch_table` can represent are emitted as replacement engine patches. Layer placement describes the desired schema; diff --git a/cmd/hclexp/decompose.go b/cmd/hclexp/decompose.go index a3c594f..ca4da34 100644 --- a/cmd/hclexp/decompose.go +++ b/cmd/hclexp/decompose.go @@ -327,12 +327,28 @@ func buildDecomposition(snapshots []decomposeSnapshot, envs []string, assignment return generatedDecomposition{}, fmt.Errorf("assignment %q requests group %q in environments [%s], but object is present in [%s]", object.key(), group.Name, strings.Join(group.Envs, ", "), strings.Join(present, ", ")) } - if !uniform { + if !uniform && (object.Kind != hclload.KindTable || !tableObjectUniformIgnoringSettings(object, group.Envs, byEnvRole)) { return generatedDecomposition{}, fmt.Errorf("assignment %q requests group %q, but object differs between member environments: %s", object.key(), group.Name, strings.Join(group.Envs, ", ")) } baseObject := objectSchema(byEnvRole[group.Envs[0]][object.Role], object) + patches := map[string]func(*hclload.Schema){} + if object.Kind == hclload.KindTable { + setTableSettingsIntersection(baseObject, object, group.Envs, byEnvRole) + for _, env := range group.Envs { + apply, err := objectPatch(baseObject, objectSchema(byEnvRole[env][object.Role], object), object) + if err != nil { + return generatedDecomposition{}, fmt.Errorf("assignment %q cannot synthesize group %q table settings base: %w", object.key(), group.Name, err) + } + patches[env] = apply + } + } addObjectToSchema(layer(layers, groupLayerPath(group.Name, object.Role)), baseObject, object) + for env, apply := range patches { + if apply != nil { + apply(layer(layers, envLayerPath(env, object.Role))) + } + } continue } roleEnvs := environmentsForRole(object.Role, envs, byEnvRole) @@ -349,6 +365,9 @@ func buildDecomposition(snapshots []decomposeSnapshot, envs []string, assignment baseline := chooseBaseline(assignment.BaselineEnv, present) baseObject := objectSchema(byEnvRole[baseline][object.Role], object) + if object.Kind == hclload.KindTable { + setTableSettingsIntersection(baseObject, object, present, byEnvRole) + } if uniform { addObjectToSchema(layer(layers, sharedLayerPath(object.Role)), baseObject, object) continue @@ -357,7 +376,7 @@ func buildDecomposition(snapshots []decomposeSnapshot, envs []string, assignment patchable := true var patchErr error for _, env := range present { - if env == baseline { + if env == baseline && object.Kind != hclload.KindTable { continue } apply, err := objectPatch(baseObject, objectSchema(byEnvRole[env][object.Role], object), object) @@ -1065,6 +1084,44 @@ func objectUniform(object decomposeObject, envs []string, schemas map[string]map return true } +func tableObjectUniformIgnoringSettings(object decomposeObject, envs []string, schemas map[string]map[string]*hclload.Schema) bool { + if len(envs) < 2 { + return true + } + withoutSettings := func(env string) *hclload.Schema { + isolated := objectSchema(schemas[env][object.Role], object) + onlyTable(isolated).Settings = nil + return isolated + } + first := withoutSettings(envs[0]) + for _, env := range envs[1:] { + equal, err := canonicalSchemasEqual(first, withoutSettings(env)) + if err != nil || !equal { + return false + } + } + return true +} + +func setTableSettingsIntersection(base *hclload.Schema, object decomposeObject, envs []string, schemas map[string]map[string]*hclload.Schema) { + intersection := map[string]string{} + for key, value := range onlyTable(objectSchema(schemas[envs[0]][object.Role], object)).Settings { + intersection[key] = value + } + for _, env := range envs[1:] { + settings := onlyTable(objectSchema(schemas[env][object.Role], object)).Settings + for key, value := range intersection { + if candidate, ok := settings[key]; !ok || candidate != value { + delete(intersection, key) + } + } + } + if len(intersection) == 0 { + intersection = nil + } + onlyTable(base).Settings = intersection +} + func canonicalSchemasEqual(a, b *hclload.Schema) (bool, error) { var left, right bytes.Buffer if err := hclload.Write(&left, a); err != nil { diff --git a/cmd/hclexp/decompose_test.go b/cmd/hclexp/decompose_test.go index 8da3ecf..0b9d76e 100644 --- a/cmd/hclexp/decompose_test.go +++ b/cmd/hclexp/decompose_test.go @@ -231,19 +231,7 @@ database "posthog" { } }`), 0o600)) out := t.TempDir() - cmd := exec.Command(os.Args[0], "-test.run=^TestDecomposeCLIProcess$", "--", - "-dump-root", dumpRoot, - "-env", "dev,prod-eu,prod-us", - "-glob", "*-ops.hcl", - "-zk-paths", "keep", - "-assignment", assignmentPath, - "-out", out, - ) - cmd.Env = append(os.Environ(), "HCLEXP_DECOMPOSE_HELPER=1") - output, err := cmd.CombinedOutput() - require.NoError(t, err, string(output)) - assert.Contains(t, string(output), "decomposed 3 environments") - assert.Contains(t, string(output), "round-trip verified") + runDecomposeCLIForTest(t, dumpRoot, "*-ops.hcl", assignmentPath, out) sharedBody, err := os.ReadFile(filepath.Join(out, sharedLayerPath("ops"))) require.NoError(t, err) @@ -270,6 +258,123 @@ database "posthog" { assert.Contains(t, manifest, `env "prod-us" { layers = ["layers/shared/ops", "layers/group/prod/ops"] }`) } +func TestDecompose_SettingsIntersection_EndToEnd(t *testing.T) { + dumpRoot := t.TempDir() + writeDump := func(env string) { + t.Helper() + residue := ` + replica_variant = "west" + storage_policy = "s3_tiered"` + if env == "prod-eu" { + residue = ` + max_replicated_merges_in_queue = "6" + parts_to_delay_insert = "1000" + parts_to_throw_insert = "3000" + prefer_fetch_merged_part_size_threshold = "1" + prefer_fetch_merged_part_time_threshold = "60" + replica_variant = "eu"` + } + settings := ` + index_granularity = "8192" + object_serialization_version = "v3" + object_shared_data_serialization_version = "map_with_buckets"` + residue + table := func(name string) string { + return ` + table "` + name + `" { + order_by = ["id"] + column "id" { type = "UInt64" } + engine "merge_tree" {} + settings = {` + settings + ` + } + }` + } + body := `node "` + env + `-ops" { + macros = { hostClusterRole = "ops" } +} +database "posthog" {` + table("sharded_query_log_archive") + if env != "dev" { + body += table("prod_query_log_archive") + } + body += "\n}" + dir := filepath.Join(dumpRoot, env) + require.NoError(t, os.MkdirAll(dir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, env+"-ops.hcl"), []byte(body), 0o600)) + } + writeDump("dev") + writeDump("prod-eu") + writeDump("prod-us") + + for _, mode := range []string{"auto", "shared"} { + t.Run(mode, func(t *testing.T) { + sharedAssignment := "" + if mode == "shared" { + sharedAssignment = ` + "ops/posthog/table/sharded_query_log_archive": { "mode": "shared" },` + } + assignmentPath := filepath.Join(t.TempDir(), "assignment.json") + require.NoError(t, os.WriteFile(assignmentPath, []byte(`{ + "version": 1, + "objects": {`+sharedAssignment+` + "ops/posthog/table/prod_query_log_archive": { + "mode": "group", + "envs": ["prod-eu", "prod-us"] + } + } +}`), 0o600)) + out := t.TempDir() + runDecomposeCLIForTest(t, dumpRoot, "*-ops.hcl", assignmentPath, out) + + read := func(path string) string { + t.Helper() + body, err := os.ReadFile(filepath.Join(out, path)) + require.NoError(t, err) + return string(body) + } + shared := read(sharedLayerPath("ops")) + assert.Contains(t, shared, `table "sharded_query_log_archive"`) + assert.Contains(t, shared, `index_granularity = "8192"`) + assert.Contains(t, shared, `object_serialization_version = "v3"`) + assert.Contains(t, shared, `object_shared_data_serialization_version = "map_with_buckets"`) + assert.NotContains(t, shared, "storage_policy") + assert.NotContains(t, shared, "max_replicated_merges_in_queue") + assert.NotContains(t, shared, "replica_variant") + + group := read(groupLayerPath("prod", "ops")) + assert.Contains(t, group, `table "prod_query_log_archive"`) + assert.Contains(t, group, `index_granularity = "8192"`) + assert.NotContains(t, group, "storage_policy") + assert.NotContains(t, group, "max_replicated_merges_in_queue") + assert.NotContains(t, group, "replica_variant") + + dev := read(envLayerPath("dev", "ops")) + assert.Contains(t, dev, `patch_table "sharded_query_log_archive"`) + assert.Contains(t, dev, `replica_variant = "west"`) + assert.Contains(t, dev, "storage_policy") + assert.Contains(t, dev, `"s3_tiered"`) + assert.NotContains(t, dev, `patch_table "prod_query_log_archive"`) + + prodUS := read(envLayerPath("prod-us", "ops")) + assert.Contains(t, prodUS, `patch_table "sharded_query_log_archive"`) + assert.Contains(t, prodUS, `patch_table "prod_query_log_archive"`) + assert.Contains(t, prodUS, `replica_variant = "west"`) + assert.Contains(t, prodUS, "storage_policy") + assert.Contains(t, prodUS, `"s3_tiered"`) + + prodEU := read(envLayerPath("prod-eu", "ops")) + assert.Contains(t, prodEU, `patch_table "sharded_query_log_archive"`) + assert.Contains(t, prodEU, `patch_table "prod_query_log_archive"`) + assert.Contains(t, prodEU, `replica_variant = "eu"`) + assert.Contains(t, prodEU, `max_replicated_merges_in_queue = "6"`) + assert.NotContains(t, prodEU, "storage_policy") + + manifest := read("manifest.hcl") + assert.Contains(t, manifest, `env "dev" { layers = ["layers/shared/ops", "layers/env/dev/ops"] }`) + assert.Contains(t, manifest, `env "prod-eu" { layers = ["layers/shared/ops", "layers/group/prod/ops", "layers/env/prod-eu/ops"] }`) + assert.Contains(t, manifest, `env "prod-us" { layers = ["layers/shared/ops", "layers/group/prod/ops", "layers/env/prod-us/ops"] }`) + }) + } +} + func TestDecomposeCLIProcess(t *testing.T) { if os.Getenv("HCLEXP_DECOMPOSE_HELPER") != "1" { return @@ -283,6 +388,23 @@ func TestDecomposeCLIProcess(t *testing.T) { t.Fatal("missing decompose CLI arguments") } +func runDecomposeCLIForTest(t *testing.T, dumpRoot, glob, assignmentPath, out string) { + t.Helper() + cmd := exec.Command(os.Args[0], "-test.run=^TestDecomposeCLIProcess$", "--", + "-dump-root", dumpRoot, + "-env", "dev,prod-eu,prod-us", + "-glob", glob, + "-zk-paths", "keep", + "-assignment", assignmentPath, + "-out", out, + ) + cmd.Env = append(os.Environ(), "HCLEXP_DECOMPOSE_HELPER=1") + output, err := cmd.CombinedOutput() + require.NoError(t, err, string(output)) + assert.Contains(t, string(output), "decomposed 3 environments") + assert.Contains(t, string(output), "round-trip verified") +} + func TestBuildDecomposition_GroupAssignmentFailsClosed(t *testing.T) { base := decomposeTable(hclload.ColumnSpec{Name: "id", Type: "UInt64"}) different := decomposeTable(