From 3aaaf93c34a1e818d57bcaf5101ca88d5ea67bd6 Mon Sep 17 00:00:00 2001 From: Ilya Kuznetsov Date: Sun, 2 Aug 2026 19:53:36 +0000 Subject: [PATCH] bundle: route config sync edits to the block that defines them Co-authored-by: Isaac --- .../split/keyed_edit/databricks.yml.tmpl | 30 ++ .../split/keyed_edit/out.test.toml | 4 + .../split/keyed_edit/output.txt | 53 +++ .../split/keyed_edit/script | 53 +++ .../split/keyed_twoblock/databricks.yml.tmpl | 39 ++ .../split/keyed_twoblock/out.test.toml | 4 + .../split/keyed_twoblock/output.txt | 64 +++ .../split/keyed_twoblock/script | 57 +++ .../split/multifile/databricks.yml.tmpl | 17 + .../split/multifile/out.test.toml | 4 + .../split/multifile/output.txt | 40 ++ .../multifile/overrides/10-first.yml.tmpl | 10 + .../multifile/overrides/20-second.yml.tmpl | 9 + .../config-remote-sync/split/multifile/script | 45 ++ .../split/positional/databricks.yml.tmpl | 30 ++ .../split/positional/out.test.toml | 4 + .../split/positional/output.txt | 68 +++ .../split/positional/script | 53 +++ .../bundle/config-remote-sync/split/test.toml | 10 + bundle/configsync/blockindex.go | 418 ++++++++++++++++++ bundle/configsync/resolve.go | 180 +++++--- bundle/configsync/resolve_test.go | 75 +--- 22 files changed, 1159 insertions(+), 108 deletions(-) create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_edit/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_edit/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_edit/output.txt create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_edit/script create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_twoblock/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_twoblock/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_twoblock/output.txt create mode 100644 acceptance/bundle/config-remote-sync/split/keyed_twoblock/script create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/output.txt create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/overrides/10-first.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/overrides/20-second.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/multifile/script create mode 100644 acceptance/bundle/config-remote-sync/split/positional/databricks.yml.tmpl create mode 100644 acceptance/bundle/config-remote-sync/split/positional/out.test.toml create mode 100644 acceptance/bundle/config-remote-sync/split/positional/output.txt create mode 100644 acceptance/bundle/config-remote-sync/split/positional/script create mode 100644 acceptance/bundle/config-remote-sync/split/test.toml create mode 100644 bundle/configsync/blockindex.go diff --git a/acceptance/bundle/config-remote-sync/split/keyed_edit/databricks.yml.tmpl b/acceptance/bundle/config-remote-sync/split/keyed_edit/databricks.yml.tmpl new file mode 100644 index 00000000000..85c310e6d50 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/keyed_edit/databricks.yml.tmpl @@ -0,0 +1,30 @@ +bundle: + name: test-bundle-$UNIQUE_NAME + +# The task list of split_job is defined in two physical blocks: the top-level +# one below and the targets.dev override further down. After loading they merge +# into one list sorted by task_key, so the merged order is +# [alpha (target), mu (top-level), zeta (top-level)] +# while each physical block keeps its own order. Editing a task must therefore +# resolve to (file, block, index-within-that-block), not to the merged index. +resources: + jobs: + split_job: + tasks: + - task_key: zeta + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/zeta + - task_key: mu + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/mu + +targets: + dev: + mode: development + resources: + jobs: + split_job: + tasks: + - task_key: alpha + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/alpha diff --git a/acceptance/bundle/config-remote-sync/split/keyed_edit/out.test.toml b/acceptance/bundle/config-remote-sync/split/keyed_edit/out.test.toml new file mode 100644 index 00000000000..4b5914daa2c --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/keyed_edit/out.test.toml @@ -0,0 +1,4 @@ +Local = true +Cloud = false +GOOS.windows = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct", "terraform"] diff --git a/acceptance/bundle/config-remote-sync/split/keyed_edit/output.txt b/acceptance/bundle/config-remote-sync/split/keyed_edit/output.txt new file mode 100644 index 00000000000..7bff6aadeca --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/keyed_edit/output.txt @@ -0,0 +1,53 @@ +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/dev/files... +Deploying resources... +Updating deployment state... +Deployment complete! + +=== Edit alpha, defined only in the target block +=== Sync +Detected changes in 1 resource(s): + +Resource: resources.jobs.split_job + tasks[task_key='alpha'].timeout_seconds: add + + + +=== Only alpha in the target block gains timeout_seconds: 111 + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -29,2 +29,3 @@ + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/alpha ++ timeout_seconds: 111 + +=== Edit mu, a top-level task that is not first after the merge sort +=== Sync +Detected changes in 1 resource(s): + +Resource: resources.jobs.split_job + tasks[task_key='mu'].timeout_seconds: add + + + +=== Only mu in the top-level block gains timeout_seconds: 222 + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -18,4 +18,5 @@ + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/mu ++ timeout_seconds: 222 + + targets: + +>>> [CLI] bundle destroy --auto-approve -t dev +The following resources will be deleted: + delete resources.jobs.split_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/dev + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/config-remote-sync/split/keyed_edit/script b/acceptance/bundle/config-remote-sync/split/keyed_edit/script new file mode 100644 index 00000000000..dcfc1028455 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/keyed_edit/script @@ -0,0 +1,53 @@ +#!/bin/bash + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve -t dev +} +trap cleanup EXIT + +$CLI bundle deploy -t dev +job_id="$(read_id.py split_job)" + + +# alpha is the only task in the target block, so it must be written at index 0 of +# that block. Its merged index is also 0, but the top-level block's index 0 is +# zeta -- a naive merged-index write lands on zeta instead. +title "Edit alpha, defined only in the target block" +edit_resource.py jobs $job_id <>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -38,2 +38,3 @@ + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/omega ++ timeout_seconds: 303 + +=== Edit a field per block, a field defined in both, and add a new field +=== Sync +Detected changes in 1 resource(s): + +Resource: resources.jobs.twoblock_job + tasks[task_key='shared'].max_retries: replace + tasks[task_key='shared'].min_retry_interval_millis: add + tasks[task_key='shared'].timeout_seconds: replace + + + +=== Each edit lands on the definition that was deployed + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -21,4 +21,5 @@ + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/shared ++ min_retry_interval_millis: 5000 + - task_key: nu + notebook_task: +@@ -33,6 +34,6 @@ + tasks: + - task_key: shared +- max_retries: 2 +- timeout_seconds: 60 ++ max_retries: 9 ++ timeout_seconds: 900 + - task_key: omega + notebook_task: + +>>> [CLI] bundle destroy --auto-approve -t dev +The following resources will be deleted: + delete resources.jobs.twoblock_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/dev + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/config-remote-sync/split/keyed_twoblock/script b/acceptance/bundle/config-remote-sync/split/keyed_twoblock/script new file mode 100644 index 00000000000..2b7d8cf82ce --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/keyed_twoblock/script @@ -0,0 +1,57 @@ +#!/bin/bash + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve -t dev +} +trap cleanup EXIT + +$CLI bundle deploy -t dev +job_id="$(read_id.py twoblock_job)" + + +# omega is defined only in the target block, but its sibling "shared" contributes +# to both blocks. Editing omega must land on omega and must leave the two-block +# element completely untouched. +title "Edit omega, whose sibling is defined in two blocks" +edit_resource.py jobs $job_id <>> diff.py databricks.yml.backup databricks.yml + +=== overrides/10-first.yml is untouched + +>>> diff.py overrides/10-first.yml.backup overrides/10-first.yml + +=== overrides/20-second.yml gains timeout_seconds: 777 + +>>> diff.py overrides/20-second.yml.backup overrides/20-second.yml +--- overrides/20-second.yml.backup ++++ overrides/20-second.yml +@@ -8,2 +8,3 @@ + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/second ++ timeout_seconds: 777 + +>>> [CLI] bundle destroy --auto-approve -t dev +The following resources will be deleted: + delete resources.jobs.multifile_job + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/dev + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/config-remote-sync/split/multifile/overrides/10-first.yml.tmpl b/acceptance/bundle/config-remote-sync/split/multifile/overrides/10-first.yml.tmpl new file mode 100644 index 00000000000..e0d765810fb --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/multifile/overrides/10-first.yml.tmpl @@ -0,0 +1,10 @@ +targets: + dev: + mode: development + resources: + jobs: + multifile_job: + tasks: + - task_key: from_first_file + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/first diff --git a/acceptance/bundle/config-remote-sync/split/multifile/overrides/20-second.yml.tmpl b/acceptance/bundle/config-remote-sync/split/multifile/overrides/20-second.yml.tmpl new file mode 100644 index 00000000000..d0e7406aa90 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/multifile/overrides/20-second.yml.tmpl @@ -0,0 +1,9 @@ +targets: + dev: + resources: + jobs: + multifile_job: + tasks: + - task_key: from_second_file + notebook_task: + notebook_path: /Users/{{workspace_user_name}}/second diff --git a/acceptance/bundle/config-remote-sync/split/multifile/script b/acceptance/bundle/config-remote-sync/split/multifile/script new file mode 100644 index 00000000000..cfef6aea3cc --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/multifile/script @@ -0,0 +1,45 @@ +#!/bin/bash + +envsubst < databricks.yml.tmpl > databricks.yml +envsubst < overrides/10-first.yml.tmpl > overrides/10-first.yml +envsubst < overrides/20-second.yml.tmpl > overrides/20-second.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve -t dev +} +trap cleanup EXIT + +$CLI bundle deploy -t dev +job_id="$(read_id.py multifile_job)" + + +# from_second_file is defined in overrides/20-second.yml. The edit must be written +# to that file only -- not to the top-level databricks.yml and not to the sibling +# override file, and without creating a duplicate subtree anywhere. +title "Edit a task defined in the second included file" +edit_resource.py jobs $job_id <>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -15,5 +15,5 @@ + clusters: + - label: default +- num_workers: 2 ++ num_workers: 5 + libraries: + - notebook: +@@ -28,3 +28,3 @@ + clusters: + - label: maintenance +- num_workers: 1 ++ num_workers: 3 + +=== Remove the target-block cluster (length change) +=== Sync +Detected changes in 1 resource(s): + +Resource: resources.pipelines.split_pipeline + clusters: replace + clusters[1].num_workers: remove + + + +=== Applied correctly or left unchanged; default cluster never corrupted + +>>> diff.py databricks.yml.backup databricks.yml +--- databricks.yml.backup ++++ databricks.yml +@@ -28,3 +28,2 @@ + clusters: + - label: maintenance +- num_workers: 3 + +>>> grep -c label: default databricks.yml +1 + +>>> [CLI] bundle destroy --auto-approve -t dev +The following resources will be deleted: + delete resources.pipelines.split_pipeline + +This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the +Streaming Tables (STs) and Materialized Views (MVs) managed by them: + delete resources.pipelines.split_pipeline + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-bundle-[UNIQUE_NAME]/dev + +Deleting files... +Destroy complete! diff --git a/acceptance/bundle/config-remote-sync/split/positional/script b/acceptance/bundle/config-remote-sync/split/positional/script new file mode 100644 index 00000000000..a1ad6715b29 --- /dev/null +++ b/acceptance/bundle/config-remote-sync/split/positional/script @@ -0,0 +1,53 @@ +#!/bin/bash + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve -t dev +} +trap cleanup EXIT + +$CLI bundle deploy -t dev +pipeline_id="$(read_id.py split_pipeline)" + + +# Length-preserving field edits on both elements of a positional list that is +# split across two blocks. Each edit must land in its own block. +title "Edit num_workers on the top-level cluster and on the target cluster" +edit_resource.py pipelines $pipeline_id <.. block and the +// targets..resources... override block, either of +// which may live in its own included file. Loading merges those regions into one +// sequence, and keyed sequences are also sorted by key, so an index into the +// merged sequence does not address any single region. Write-back has to map a +// merged position back to a region and to the position inside that region. +type sourceBlock struct { + // prefix is empty for a top-level block and "targets." for a target + // override block. Both may occur in the same file, so the file alone does not + // identify a block. + prefix string + file string +} + +// blockResolver answers which physical block a merged value came from. +// +// It works by location: a merged value keeps the locations of the entries it was +// loaded from, and merging accumulates them (see libs/dyn/merge), so a value +// assembled from two blocks reports a location in each. Selecting a target folds +// its overrides into the resources tree and drops the targets subtree, so the +// blocks are recovered by parsing the contributing files again, where the two +// regions are still separate. +type blockResolver struct { + // blocks holds one parsed file per contributing file, keyed by block, so a + // path relative to a block can be looked up inside it. + blocks map[sourceBlock]dyn.Value + target string + byLocation map[dyn.Location]sourceBlock +} + +// newBlockResolver builds the location -> block mapping for the bundle's +// resources. It returns nil when the source cannot be re-read, in which case +// callers keep their existing single-block behaviour. +// +// Only the files the merged configuration actually references are read, and they +// are parsed directly rather than reloaded through the mutator pipeline: the +// pipeline resolves includes, reports through logdiag and executes the bundle's +// preinit script, none of which belongs to reading back a source location. +func newBlockResolver(ctx context.Context, b *bundle.Bundle) *blockResolver { + root := b.Config.Value() + if !root.IsValid() { + return nil + } + + r := &blockResolver{ + blocks: make(map[sourceBlock]dyn.Value), + target: b.Config.Bundle.Target, + byLocation: make(map[dyn.Location]sourceBlock), + } + + for _, file := range slices.Sorted(maps.Keys(referencedFiles(root))) { + contents, err := os.ReadFile(file) + if err != nil { + log.Debugf(ctx, "config-remote-sync: cannot read %s, treating its sequences as unsplit: %v", file, err) + continue + } + parsed, diags := config.LoadFromBytes(file, contents) + if diags.HasError() { + log.Debugf(ctx, "config-remote-sync: cannot parse %s, treating its sequences as unsplit: %v", file, diags.Error()) + continue + } + r.indexRegion(parsed.Value(), sourceBlock{file: file}) + if r.target != "" { + r.indexRegion(parsed.Value(), sourceBlock{prefix: "targets." + r.target, file: file}) + } + } + + if len(r.byLocation) == 0 { + return nil + } + return r +} + +// referencedFiles returns the files the merged configuration was loaded from. +func referencedFiles(root dyn.Value) map[string]struct{} { + files := map[string]struct{}{} + _ = dyn.WalkReadOnly(root, func(_ dyn.Path, v dyn.Value) error { + for _, location := range v.Locations() { + if location.File != "" { + files[location.File] = struct{}{} + } + } + return nil + }) + return files +} + +// indexRegion records every location under one region's resources subtree of a +// parsed file, and keeps the parsed file so paths can be resolved inside it. +func (r *blockResolver) indexRegion(parsed dyn.Value, block sourceBlock) { + subtree, err := dyn.GetByPath(parsed, r.regionPath(block.prefix, dyn.NewPath(dyn.Key("resources")))) + if err != nil { + return + } + r.blocks[block] = parsed + _ = dyn.WalkReadOnly(subtree, func(_ dyn.Path, v dyn.Value) error { + for _, location := range v.Locations() { + // A file only carries locations of its own, so a location seen here + // belongs to this block. First writer wins: an outer node accumulates + // its children's locations, but the innermost node that owns a location + // is the one walked last. + if location.File != block.file { + continue + } + if _, ok := r.byLocation[location]; !ok { + r.byLocation[location] = block + } + } + return nil + }) +} + +// regionPath prefixes a resources-relative path with the region it belongs to. +func (r *blockResolver) regionPath(prefix string, path dyn.Path) dyn.Path { + if prefix == "" { + return path + } + return append(dyn.NewPath(dyn.Key("targets"), dyn.Key(r.target)), path...) +} + +// sortedBlocks lists the known blocks with the top-level ones first and a total +// order within each scope, so a choice between blocks never depends on map or +// location iteration order. +func (r *blockResolver) sortedBlocks() []sourceBlock { + blocks := slices.Collect(maps.Keys(r.blocks)) + slices.SortFunc(blocks, compareBlocks) + return blocks +} + +// compareBlocks orders top-level blocks before target blocks, then by file. +func compareBlocks(a, b sourceBlock) int { + if (a.prefix == "") != (b.prefix == "") { + if a.prefix == "" { + return -1 + } + return 1 + } + return cmp.Or(cmp.Compare(a.prefix, b.prefix), cmp.Compare(a.file, b.file)) +} + +// blocksOf returns the distinct blocks that contributed to value, sorted with the +// top-level block first. More than one result means the value is assembled from +// several regions and has no single source location. +func (r *blockResolver) blocksOf(value dyn.Value) []sourceBlock { + var blocks []sourceBlock + for _, location := range value.Locations() { + block, ok := r.byLocation[location] + if !ok { + continue + } + if !slices.Contains(blocks, block) { + blocks = append(blocks, block) + } + } + // Order is total and independent of how locations happened to accumulate, so + // callers that prefer the top-level block get a deterministic answer. + slices.SortFunc(blocks, compareBlocks) + return blocks +} + +// winningBlock returns the block whose definition of value is the one the merged +// configuration ended up with. +// +// A scalar defined in several blocks is not ambiguous: merging keeps the incoming +// value and records its location first (see mergePrimitive in libs/dyn/merge), so +// the deployed value came from whichever block owns Locations()[0]. Writing the +// remote change anywhere else would leave the effective value unchanged. +func (r *blockResolver) winningBlock(value dyn.Value) (sourceBlock, bool) { + for _, location := range value.Locations() { + if block, ok := r.byLocation[location]; ok { + return block, true + } + } + return sourceBlock{}, false +} + +// localIndex returns the position of element inside the sequence that block +// defines at sequencePath, where sequencePath is relative to the block. +func (r *blockResolver) localIndex(block sourceBlock, sequencePath dyn.Path, element dyn.Value) (int, bool) { + parsed, ok := r.blocks[block] + if !ok { + return 0, false + } + sequence, err := dyn.GetByPath(parsed, r.regionPath(block.prefix, sequencePath)) + if err != nil { + return 0, false + } + entries, ok := sequence.AsSequence() + if !ok { + return 0, false + } + + locations := make(map[dyn.Location]struct{}, len(element.Locations())) + for _, location := range element.Locations() { + locations[location] = struct{}{} + } + + // A region's sequence is the concatenation of the entries contributed by each + // included file, but the write targets one file, so the index has to be + // counted among that file's entries only. + local := 0 + for _, entry := range entries { + entryFile := entry.Location().File + if entryFile != block.file { + continue + } + for _, location := range entry.Locations() { + if _, ok := locations[location]; ok { + return local, true + } + } + local++ + } + return 0, false +} + +// errAmbiguousBlock reports a change that cannot be attributed to one physical +// block. Leaving such a change unapplied is preferable to writing it to a guessed +// location, since the sync runs unattended and a later run can retry. +var errAmbiguousBlock = errors.New("change cannot be attributed to a single source block") + +// route maps a change resolved against the merged configuration onto the physical +// block that owns it, rewriting merged sequence indices into block-local ones. +// +// The element that the change addresses decides the block: an element defined in +// exactly one block is written there. When the addressed element is assembled from +// several blocks, the leaf field decides instead, because merging records the +// location of each field separately. If neither identifies a single block the +// change is ambiguous and is reported as such. +func (r *blockResolver) route(change resolvedChange) (sourceBlock, *structpath.PatternNode, error) { + block, err := r.blockFor(change) + if err != nil { + return sourceBlock{}, nil, err + } + + path, err := r.localize(block, change) + if err != nil { + return sourceBlock{}, nil, err + } + return block, path, nil +} + +// blockFor picks the block a change belongs to. +func (r *blockResolver) blockFor(change resolvedChange) (sourceBlock, error) { + // A new element has no source of its own; it is placed relative to the + // sequence that receives it. + for _, step := range change.steps { + if step.newElement { + return r.blockForNewElement(change) + } + } + + if len(change.steps) > 0 { + last := change.steps[len(change.steps)-1] + blocks := r.blocksOf(last.element) + switch len(blocks) { + case 1: + return blocks[0], nil + case 0: + return sourceBlock{}, fmt.Errorf("%w: no source location for the addressed element", errAmbiguousBlock) + } + + // The element is assembled from several blocks, but a field of it is not: + // it is written in one block, or written in both and one of them wins. A + // field-level change therefore routes by the field's own location. + if change.leaf.IsValid() { + if block, ok := r.winningBlock(change.leaf); ok { + return block, nil + } + return sourceBlock{}, fmt.Errorf("%w: element is defined in %d blocks and the field has no source location", errAmbiguousBlock, len(blocks)) + } + + // A field that exists in no block is new. It has no location to route by, + // but it still needs a defined destination, so it goes to the block that + // declares the resource rather than a target-specific scope. + if len(change.path.AsSlice()) > last.component+1 { + for _, block := range blocks { + if block.prefix == "" { + return block, nil + } + } + return blocks[0], nil + } + + // The change addresses the element itself, not one of its fields. Removing + // or recreating it cannot be expressed against a single block. + return sourceBlock{}, fmt.Errorf("%w: element is defined in %d blocks", errAmbiguousBlock, len(blocks)) + } + + // No sequence on the path: route by the leaf's own location. + if change.leaf.IsValid() { + if block, ok := r.winningBlock(change.leaf); ok { + return block, nil + } + } + return sourceBlock{}, fmt.Errorf("%w: no source location for the change", errAmbiguousBlock) +} + +// blockForNewElement chooses where an element that exists in no source block +// should be written: the block that defines the sequence, or when several do, the +// block that declares the resource. Adding to the resource's own block keeps a new +// element out of a target-specific scope the user did not ask for. +func (r *blockResolver) blockForNewElement(change resolvedChange) (sourceBlock, error) { + step := change.steps[len(change.steps)-1] + + // A nested sequence is reached through the enclosing elements, which are + // already placed: route by the innermost one that exists in the source, since + // a new element belongs in the same block as its parent. + for i := len(change.steps) - 2; i >= 0; i-- { + enclosing := change.steps[i] + if enclosing.newElement { + continue + } + if blocks := r.blocksOf(enclosing.element); len(blocks) == 1 { + return blocks[0], nil + } + } + + var blocks []sourceBlock + for _, block := range r.sortedBlocks() { + sequence, err := dyn.GetByPath(r.blocks[block], r.regionPath(block.prefix, step.sequencePath)) + if err != nil { + continue + } + for _, location := range sequence.Locations() { + if location.File == block.file && !slices.Contains(blocks, block) { + blocks = append(blocks, block) + } + } + } + + switch len(blocks) { + case 0: + return sourceBlock{}, fmt.Errorf("%w: no block defines the sequence receiving the new element", errAmbiguousBlock) + case 1: + return blocks[0], nil + } + + // blocksOf orders the top-level block first, and the resource is declared + // there whenever a top-level block exists at all. + for _, block := range blocks { + if block.prefix == "" { + return block, nil + } + } + return blocks[0], nil +} + +// localize rewrites the merged sequence indices in a change's path into indices +// inside block. +func (r *blockResolver) localize(block sourceBlock, change resolvedChange) (*structpath.PatternNode, error) { + nodes := change.path.AsSlice() + local := make([]int, len(change.steps)) + for i, step := range change.steps { + if step.newElement { + // Keeps the [*] placeholder; the patcher appends to the block's + // sequence. + local[i] = -1 + continue + } + // A nested sequence is reached through the outer sequences on the path, + // whose indices refer to the merged view. Rewrite them to the block's own + // indices, otherwise the lookup addresses the wrong parent element. + sequencePath := slices.Clone(step.sequencePath) + for j := range i { + outer := change.steps[j] + if local[j] < 0 || len(outer.sequencePath) >= len(sequencePath) { + continue + } + sequencePath[len(outer.sequencePath)] = dyn.Index(local[j]) + } + + index, ok := r.localIndex(block, sequencePath, step.element) + if !ok { + return nil, fmt.Errorf("%w: element has no position in %s", errAmbiguousBlock, block.file) + } + local[i] = index + } + + var result *structpath.PatternNode + next := 0 + for component, node := range nodes { + if next < len(change.steps) && change.steps[next].component == component { + if local[next] < 0 { + result = structpath.NewPatternBracketStar(result) + } else { + result = structpath.NewPatternIndex(result, local[next]) + } + next++ + continue + } + key, ok := node.StringKey() + if !ok { + return nil, fmt.Errorf("%w: unsupported path component in %s", errAmbiguousBlock, change.path.String()) + } + result = structpath.NewPatternStringKey(result, key) + } + return result, nil +} diff --git a/bundle/configsync/resolve.go b/bundle/configsync/resolve.go index 01018365e67..38814f76189 100644 --- a/bundle/configsync/resolve.go +++ b/bundle/configsync/resolve.go @@ -3,6 +3,7 @@ package configsync import ( "cmp" "context" + "errors" "fmt" "io/fs" "maps" @@ -23,23 +24,50 @@ type FieldChange struct { FieldCandidates []string } -// resolveSelectors converts key-value selectors to numeric indices that match -// the YAML file positions. It also returns the location of the resolved leaf value. +// sequenceStep records a sequence element that a change path navigated through. +// The element's index inside a physical block is only known once the block has +// been chosen, so the merged position is kept until then. +type sequenceStep struct { + // position in the pattern built so far, i.e. how many components precede + // this sequence's index. + component int + // path of the sequence relative to a block, e.g. resources.jobs.j.tasks. + sequencePath dyn.Path + element dyn.Value + // newElement marks an Add whose key is not in the merged sequence yet. + newElement bool +} + +// resolvedChange locates a change in the source YAML. +type resolvedChange struct { + // path is relative to the block, with merged indices still in place. + path *structpath.PatternNode + steps []sequenceStep + // leaf is the merged value the change addresses, invalid for a new field. + leaf dyn.Value +} + +// resolveSelectors converts key-value selectors to the indices of the merged +// configuration and records the sequence elements traversed on the way, so the +// caller can map those positions onto the physical block that owns them. // Example: "resources.jobs.foo.tasks[task_key='main'].name" -> "resources.jobs.foo.tasks[1].name" // Returns a PatternNode because for Add operations, [*] may be used as a placeholder for new elements. -func resolveSelectors(pathStr string, b *bundle.Bundle, operation OperationType) (*structpath.PatternNode, dyn.Location, error) { +func resolveSelectors(pathStr string, b *bundle.Bundle, operation OperationType) (resolvedChange, error) { node, err := structpath.ParsePath(pathStr) if err != nil { - return nil, dyn.Location{}, fmt.Errorf("failed to parse path %s: %w", pathStr, err) + return resolvedChange{}, fmt.Errorf("failed to parse path %s: %w", pathStr, err) } nodes := node.AsSlice() var result *structpath.PatternNode + var steps []sequenceStep + var currentPath dyn.Path currentValue := b.Config.Value() - for _, n := range nodes { + for component, n := range nodes { if key, ok := n.StringKey(); ok { result = structpath.NewPatternStringKey(result, key) + currentPath = append(currentPath, dyn.Key(key)) if currentValue.IsValid() { currentValue, _ = dyn.GetByPath(currentValue, dyn.Path{dyn.Key(key)}) } @@ -47,17 +75,26 @@ func resolveSelectors(pathStr string, b *bundle.Bundle, operation OperationType) } if idx, ok := n.Index(); ok { + sequencePath := slices.Clone(currentPath) result = structpath.NewPatternIndex(result, idx) + currentPath = append(currentPath, dyn.Index(idx)) + var element dyn.Value if currentValue.IsValid() { - currentValue, _ = dyn.GetByPath(currentValue, dyn.Path{dyn.Index(idx)}) + element, _ = dyn.GetByPath(currentValue, dyn.Path{dyn.Index(idx)}) + currentValue = element } + steps = append(steps, sequenceStep{ + component: component, + sequencePath: sequencePath, + element: element, + }) continue } // Check for key-value selector: [key='value'] if key, value, ok := n.KeyValue(); ok { if !currentValue.IsValid() || currentValue.Kind() != dyn.KindSequence { - return nil, dyn.Location{}, fmt.Errorf("cannot apply [%s='%s'] selector to non-array value in path %s", key, value, pathStr) + return resolvedChange{}, fmt.Errorf("cannot apply [%s='%s'] selector to non-array value in path %s", key, value, pathStr) } seq, _ := currentValue.AsSequence() @@ -75,50 +112,36 @@ func resolveSelectors(pathStr string, b *bundle.Bundle, operation OperationType) } } + sequencePath := slices.Clone(currentPath) + if foundIndex == -1 { if operation == OperationAdd { result = structpath.NewPatternBracketStar(result) + steps = append(steps, sequenceStep{ + component: component, + sequencePath: sequencePath, + newElement: true, + }) // Can't navigate further into non-existent element currentValue = dyn.Value{} continue } - return nil, dyn.Location{}, fmt.Errorf("no array element found with %s='%s' in path %s", key, value, pathStr) + return resolvedChange{}, fmt.Errorf("no array element found with %s='%s' in path %s", key, value, pathStr) } - // Mutators may reorder sequence elements (e.g., tasks sorted by task_key). - // Use location information to determine the original YAML file position. - yamlIndex := yamlFileIndex(seq, foundIndex) - result = structpath.NewPatternIndex(result, yamlIndex) + result = structpath.NewPatternIndex(result, foundIndex) + currentPath = append(currentPath, dyn.Index(foundIndex)) + steps = append(steps, sequenceStep{ + component: component, + sequencePath: sequencePath, + element: seq[foundIndex], + }) currentValue = seq[foundIndex] continue } } - return result, currentValue.Location(), nil -} - -// yamlFileIndex determines the original YAML file position of a sequence element. -// Mutators may reorder sequence elements (e.g., tasks sorted by task_key), so the -// in-memory index may not match the position in the YAML file. This function uses -// location information to count how many elements from the same file appear before -// the target element, giving the correct index for YAML patching. -func yamlFileIndex(seq []dyn.Value, sortedIndex int) int { - matchLocation := seq[sortedIndex].Location() - if matchLocation.File == "" { - return sortedIndex - } - - yamlIndex := 0 - for i, elem := range seq { - if i == sortedIndex { - continue - } - loc := elem.Location() - if loc.File == matchLocation.File && loc.Line < matchLocation.Line { - yamlIndex++ - } - } - return yamlIndex + return resolvedChange{path: result, steps: steps, leaf: currentValue}, nil } func pathDepth(pathStr string) int { @@ -132,7 +155,7 @@ func pathDepth(pathStr string) int { // adjustArrayIndex adjusts the index in a PatternNode based on previous operations. // When operations are applied sequentially, removals and additions shift array indices. // This function adjusts the index to account for those shifts. -func adjustArrayIndex(path *structpath.PatternNode, operations map[string][]struct { +func adjustArrayIndex(path *structpath.PatternNode, scope string, operations map[string][]struct { index int operation OperationType }, @@ -143,7 +166,7 @@ func adjustArrayIndex(path *structpath.PatternNode, operations map[string][]stru } parentPath := path.Parent() - parentPathStr := parentPath.String() + parentPathStr := scope + parentPath.String() ops := operations[parentPathStr] adjustment := 0 @@ -168,6 +191,7 @@ func adjustArrayIndex(path *structpath.PatternNode, operations map[string][]stru func ResolveChanges(ctx context.Context, b *bundle.Bundle, configChanges Changes) ([]FieldChange, error) { var result []FieldChange targetName := b.Config.Bundle.Target + blocks := newBlockResolver(ctx, b) resourceKeys := slices.Sorted(maps.Keys(configChanges)) @@ -215,23 +239,51 @@ func ResolveChanges(ctx context.Context, b *bundle.Bundle, configChanges Changes configChange := resourceChanges[fieldPath] fullPath := resourceKey + "." + fieldPath - resolvedPath, resolvedLocation, err := resolveSelectors(fullPath, b, configChange.Operation) + resolved, err := resolveSelectors(fullPath, b, configChange.Operation) if err != nil { return nil, fmt.Errorf("failed to resolve selectors in path %s: %w", fullPath, err) } + resolvedPath := resolved.path + + // A sequence element addressed by the merged view has to be mapped onto + // the physical block that defines it, because the merged order and the + // per-block order differ once a sequence is split across blocks. + var block sourceBlock + routed := false + if blocks != nil && len(resolved.steps) > 0 { + var routeErr error + block, resolvedPath, routeErr = blocks.route(resolved) + if routeErr != nil { + if errors.Is(routeErr, errAmbiguousBlock) { + // Applying this change would mean guessing a location. + // Leave it unapplied; a later run can pick it up. + log.Debugf(ctx, "config-remote-sync: skipping %s: %v", fullPath, routeErr) + continue + } + return nil, fmt.Errorf("failed to route change %s: %w", fullPath, routeErr) + } + routed = true + } + + // Index bookkeeping is scoped to a block so that shifts caused by + // operations on one block cannot move indices in another. + scope := "" + if routed { + scope = block.prefix + "\x00" + block.file + "\x00" + } // If the element is removed, we can use the index to replace it with added element // That may improve the diff in cases when the task is recreated because of renaming if configChange.Operation == OperationRemove { freeIndex, ok := resolvedPath.Index() if ok { - parentPath := resolvedPath.Parent().String() + parentPath := scope + resolvedPath.Parent().String() indicesToReplaceMap[parentPath] = append(indicesToReplaceMap[parentPath], freeIndex) } } if configChange.Operation == OperationAdd && resolvedPath.BracketStar() { - parentPath := resolvedPath.Parent().String() + parentPath := scope + resolvedPath.Parent().String() indices, ok := indicesToReplaceMap[parentPath] if ok && len(indices) > 0 { index := indices[0] @@ -240,11 +292,11 @@ func ResolveChanges(ctx context.Context, b *bundle.Bundle, configChanges Changes } } - resolvedPath = adjustArrayIndex(resolvedPath, indexOperations) + resolvedPath = adjustArrayIndex(resolvedPath, scope, indexOperations) // Track this operation for future index adjustments (only for array element operations) if originalIndex, ok := resolvedPath.Index(); ok { - parentPath := resolvedPath.Parent().String() + parentPath := scope + resolvedPath.Parent().String() indexOperations[parentPath] = append(indexOperations[parentPath], struct { index int operation OperationType @@ -252,15 +304,31 @@ func ResolveChanges(ctx context.Context, b *bundle.Bundle, configChanges Changes } resolvedPathStr := resolvedPath.String() - candidates := []string{resolvedPathStr} - if targetName != "" { - targetPrefixedPath := "targets." + targetName + "." + resolvedPathStr - candidates = append(candidates, targetPrefixedPath) + var candidates []string + if routed { + // The block is known, so there is exactly one path to write. + if block.prefix == "" { + candidates = []string{resolvedPathStr} + } else { + candidates = []string{block.prefix + "." + resolvedPathStr} + } + } else { + candidates = []string{resolvedPathStr} + if targetName != "" { + targetPrefixedPath := "targets." + targetName + "." + resolvedPathStr + candidates = append(candidates, targetPrefixedPath) + } } - filePath := resolvedLocation.File - + // A routed change has a known destination file even when the leaf + // itself is new, but "defined in the config" must still be decided by + // the leaf: a field with no source location is added, not replaced. + filePath := resolved.leaf.Location().File isDefinedInConfig := filePath != "" + if routed && isDefinedInConfig { + filePath = block.file + } + if !isDefinedInConfig { if configChange.Operation == OperationRemove { // If the field is not defined in the config and the operation is remove, it is more likely a CLI default @@ -274,13 +342,19 @@ func ResolveChanges(ctx context.Context, b *bundle.Bundle, configChanges Changes configChange.Operation = OperationAdd } - resourceLocation := b.Config.GetLocation(resourceKey) - filePath = resourceLocation.File + if routed { + // The enclosing element was resolved to a block, so a new field + // on it belongs in that same block. + filePath = block.file + } else { + resourceLocation := b.Config.GetLocation(resourceKey) + filePath = resourceLocation.File + } if filePath == "" { return nil, fmt.Errorf("failed to find location for resource %s for a field %s", resourceKey, fieldPath) } - log.Debugf(ctx, "Field %s has no location, using resource location: %s", fullPath, filePath) + log.Debugf(ctx, "Field %s has no location, using %s", fullPath, filePath) } if (configChange.Operation == OperationAdd || configChange.Operation == OperationReplace) && b.SyncRootPath != "" { diff --git a/bundle/configsync/resolve_test.go b/bundle/configsync/resolve_test.go index 9264ad7f5dc..63f4231d014 100644 --- a/bundle/configsync/resolve_test.go +++ b/bundle/configsync/resolve_test.go @@ -8,12 +8,22 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config/mutator" "github.com/databricks/cli/libs/cmdio" - "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/cli/libs/structs/structpath" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// resolveSelectorsPath resolves a change path and returns only the pattern, for +// tests that do not exercise block routing. +func resolveSelectorsPath(pathStr string, b *bundle.Bundle, operation OperationType) (*structpath.PatternNode, error) { + resolved, err := resolveSelectors(pathStr, b, operation) + if err != nil { + return nil, err + } + return resolved.path, nil +} + func TestResolveSelectors_NoSelectors(t *testing.T) { ctx := logdiag.InitContext(t.Context()) tmpDir := t.TempDir() @@ -32,7 +42,7 @@ func TestResolveSelectors_NoSelectors(t *testing.T) { mutator.DefaultMutators(ctx, b) - result, _, err := resolveSelectors("resources.jobs.test_job.name", b, OperationReplace) + result, err := resolveSelectorsPath("resources.jobs.test_job.name", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.name", result.String()) } @@ -57,11 +67,11 @@ func TestResolveSelectors_NumericIndices(t *testing.T) { mutator.DefaultMutators(ctx, b) - result, _, err := resolveSelectors("resources.jobs.test_job.tasks[0].task_key", b, OperationReplace) + result, err := resolveSelectorsPath("resources.jobs.test_job.tasks[0].task_key", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.tasks[0].task_key", result.String()) - result, _, err = resolveSelectors("resources.jobs.test_job.tasks[1].task_key", b, OperationReplace) + result, err = resolveSelectorsPath("resources.jobs.test_job.tasks[1].task_key", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.tasks[1].task_key", result.String()) } @@ -90,11 +100,11 @@ func TestResolveSelectors_KeyValueSelector(t *testing.T) { mutator.DefaultMutators(ctx, b) - result, _, err := resolveSelectors("resources.jobs.test_job.tasks[task_key='main'].notebook_task.notebook_path", b, OperationReplace) + result, err := resolveSelectorsPath("resources.jobs.test_job.tasks[task_key='main'].notebook_task.notebook_path", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.tasks[1].notebook_task.notebook_path", result.String()) - result, _, err = resolveSelectors("resources.jobs.test_job.tasks[task_key='setup'].notebook_task.notebook_path", b, OperationReplace) + result, err = resolveSelectorsPath("resources.jobs.test_job.tasks[task_key='setup'].notebook_task.notebook_path", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.tasks[0].notebook_task.notebook_path", result.String()) } @@ -120,7 +130,7 @@ func TestResolveSelectors_SelectorNotFound(t *testing.T) { mutator.DefaultMutators(ctx, b) - _, _, err = resolveSelectors("resources.jobs.test_job.tasks[task_key='nonexistent'].notebook_task.notebook_path", b, OperationReplace) + _, err = resolveSelectorsPath("resources.jobs.test_job.tasks[task_key='nonexistent'].notebook_task.notebook_path", b, OperationReplace) require.Error(t, err) assert.Contains(t, err.Error(), "no array element found with task_key='nonexistent'") } @@ -143,7 +153,7 @@ func TestResolveSelectors_SelectorOnNonArray(t *testing.T) { mutator.DefaultMutators(ctx, b) - _, _, err = resolveSelectors("resources.jobs.test_job[task_key='main'].name", b, OperationReplace) + _, err = resolveSelectorsPath("resources.jobs.test_job[task_key='main'].name", b, OperationReplace) require.Error(t, err) assert.Contains(t, err.Error(), "cannot apply [task_key='main'] selector to non-array value") } @@ -174,7 +184,7 @@ func TestResolveSelectors_NestedSelectors(t *testing.T) { mutator.DefaultMutators(ctx, b) - result, _, err := resolveSelectors("resources.jobs.test_job.tasks[task_key='main'].libraries[0].pypi.package", b, OperationReplace) + result, err := resolveSelectorsPath("resources.jobs.test_job.tasks[task_key='main'].libraries[0].pypi.package", b, OperationReplace) require.NoError(t, err) assert.Equal(t, "resources.jobs.test_job.tasks[1].libraries[0].pypi.package", result.String()) } @@ -200,52 +210,7 @@ func TestResolveSelectors_WildcardNotSupported(t *testing.T) { mutator.DefaultMutators(ctx, b) - _, _, err = resolveSelectors("resources.jobs.test_job.tasks.*.task_key", b, OperationReplace) + _, err = resolveSelectorsPath("resources.jobs.test_job.tasks.*.task_key", b, OperationReplace) require.Error(t, err) assert.Contains(t, err.Error(), "wildcards not allowed in path") } - -func TestYamlFileIndex(t *testing.T) { - // Simulate a sequence that was sorted alphabetically by a mutator. - // Original YAML order: notebook_task (line 10), python_wheel_task (line 20), pipeline_task (line 30), extra (line 40) - // Sorted order: extra (line 40), notebook_task (line 10), pipeline_task (line 30), python_wheel_task (line 20) - seq := []dyn.Value{ - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 40}}), // extra - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 10}}), // notebook_task - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 30}}), // pipeline_task - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 20}}), // python_wheel_task - } - - assert.Equal(t, 3, yamlFileIndex(seq, 0)) // extra: 3 elements before it in YAML - assert.Equal(t, 0, yamlFileIndex(seq, 1)) // notebook_task: first in YAML - assert.Equal(t, 2, yamlFileIndex(seq, 2)) // pipeline_task: 2 elements before it - assert.Equal(t, 1, yamlFileIndex(seq, 3)) // python_wheel_task: 1 element before it -} - -func TestYamlFileIndex_MultipleFiles(t *testing.T) { - // Tasks from two different files, sorted alphabetically by mutator. - // File A (lines 10, 20): task_a, task_b - // File B (lines 5, 15): task_c, task_d - // Sorted order: task_a (A:10), task_b (A:20), task_c (B:5), task_d (B:15) - seq := []dyn.Value{ - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 10}}), // task_a - dyn.NewValue(nil, []dyn.Location{{File: "a.yml", Line: 20}}), // task_b - dyn.NewValue(nil, []dyn.Location{{File: "b.yml", Line: 5}}), // task_c - dyn.NewValue(nil, []dyn.Location{{File: "b.yml", Line: 15}}), // task_d - } - - // Indices are relative to each file - assert.Equal(t, 0, yamlFileIndex(seq, 0)) // task_a: first in file A - assert.Equal(t, 1, yamlFileIndex(seq, 1)) // task_b: second in file A - assert.Equal(t, 0, yamlFileIndex(seq, 2)) // task_c: first in file B - assert.Equal(t, 1, yamlFileIndex(seq, 3)) // task_d: second in file B -} - -func TestYamlFileIndex_NoLocation(t *testing.T) { - seq := []dyn.Value{ - dyn.NewValue(nil, nil), - dyn.NewValue(nil, nil), - } - assert.Equal(t, 0, yamlFileIndex(seq, 0)) - assert.Equal(t, 1, yamlFileIndex(seq, 1)) -}