Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/<name>/<role>` 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;
Expand Down
61 changes: 59 additions & 2 deletions cmd/hclexp/decompose.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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 {
Expand Down
148 changes: 135 additions & 13 deletions cmd/hclexp/decompose_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
Expand All @@ -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(
Expand Down