feat: Add caching node to dataobjscan node

[salvacorts]

What this PR does / why we need it:
We see roughly 80% of the tasks results are empty. Most of the times, the output of dataobscans are way bigger than the output of the task that contains it (because of TopK, projection and aggregation nodes), but we can still leverage an negative result cache to reuse the work of scans even if the parent node change (like different aggregations over the same scan).

This PR adds a new cache node right on top of the dataobjscans, and a new field to configure a different max cacheable size the dataobjscan cache.

Special notes for your reviewer:

There are some tasks that lack any relevant node between the scan and the batching node, which would result in both caching nodes caching the same result. E.g.

Cache max_cacheable_size=5.0 MiB hashed_key=<...> key=<...>
  └── Batching batch_size=8192
      └── Compat src=metadata dst=metadata collisions=(label)                                                                                                                                                  
          └── Cache max_cacheable_size=0 B hashed_key=<...> key=<...>
              └── DataObjScan location=<...> streams=<...> section_id=<...> projections=() predicate[0]=GTE(builtin.timestamp, 2026-03-25T07:59:00Z) predicate[1]=LT(builtin.timestamp, 2026-03-25T08:22:00Z)  
                      └── @max_time_range start=2026-03-25T07:43:47.0586196Z end=2026-03-25T08:13:46.958911328Z```

As shown in the plan above, we plan to configure the max cacheable size to be 0, so we would store only empty results.
Even though they would store the same response twice in the cache under two different keys, this should be fine since the result is empty (for which we store an empty buffer). We can add an optimizer to get rid of these redundant caches in a followup PR.

Checklist

• Reviewed the CONTRIBUTING.md guide (required)
• Documentation added
• Tests updated
• Title matches the required conventional commits format, see here
  • Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
• Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
• If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

---

Note

Medium Risk
Touches query-engine planning/execution caching paths and introduces a new cache keying/injection mechanism for DataObjScan, which could affect cache hit rates and correctness if keys/limits are misconfigured, but changes are scoped to the experimental cache feature.

Overview
Adds a second, scan-level caching layer by injecting a Cache node directly above each DataObjScan (new TaskCacheDataObjScanResult), in addition to the existing task-fragment cache, so scan results (especially empty ones) can be reused across different downstream operators.

Splits cache configuration into two independent limits (task_result_max_cacheable_size and dataobjscan_result_max_cacheable_size), wires them through engine/workflow options, and updates the executor cache registry to create a dedicated store/prefix plus separate xcap metrics for DataObjScan caching; also tightens caching pipeline behavior/logging (hashed key logging, skip caching empty batches, clearer span attr naming) and updates docs/tests accordingly.

^{Written by Cursor Bugbot for commit a6144e0. This will update automatically on new commits. Configure here.}

[github-actions]

💻 Deploy preview deleted (feat: Add caching node to dataobjscan node).

[Copilot]

Pull request overview

This PR adds a second caching layer specifically for raw DataObjScan outputs (intended for “negative result” reuse across different upstream operators), alongside the existing task-fragment result caching, and exposes separate configuration for that scan-level cache.

Changes:

• Split workflow caching configuration into task-result caching vs DataObjScan-result caching.
• Inject a Cache node directly above DataObjScan nodes during workflow planning (configurable max cacheable size; commonly set to 0 to cache only empty scan results).
• Add a new query_engine.dataobjscan_result_cache configuration block and register a new cache namespace (dataobjscan-result).

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
pkg/engine/internal/workflow/workflow_planner.go	Adds scan-level cache injection across task fragments.
pkg/engine/internal/planner/physical/cache.go	Implements WrapDataObjScansWithCache and adds new cache type + shared separator.
pkg/engine/internal/executor/caching.go	Registers a new backing cache instance for dataobjscan-result.
pkg/engine/engine.go	Adds new executor config block + toggles for scan-level caching; threads options into workflow planning.
pkg/engine/internal/workflow/workflow.go	Extends workflow.Options and passes new cache params into planning.
pkg/engine/internal/workflow/workflow_planner_test.go	Updates expected plans to include scan-level cache nodes.
pkg/engine/internal/workflow/metastore_workflow_planner_test.go	Updates caching params usage; adjusts comment.
pkg/engine/internal/planner/physical/printer.go	Prints cache max size as bytes (“0 B” etc.) rather than “unlimited”.
pkg/engine/engine_test.go	Updates buildWorkflow call signature in tests.
docs/sources/shared/configuration.md	Documents new dataobjscan_result_cache block and CLI prefixes.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Error-level log for expected passthrough code path
  • Changed the passthrough EOF log in cachingPipeline.Read from error to debug so expected non-cached non-empty results no longer emit error-level noise.
• ✅ Fixed: Inject moves all parent children, not just DataObjScan
  • Updated DataObjScan cache wrapping to remove only the parent→scan edge and then add parent→cache→scan edges, preserving any sibling children, with DAG RemoveEdge support and tests.

Or push these changes by commenting:

@cursor push 4339513524

Preview (4339513524)

diff --git a/pkg/engine/internal/executor/caching.go b/pkg/engine/internal/executor/caching.go
--- a/pkg/engine/internal/executor/caching.go
+++ b/pkg/engine/internal/executor/caching.go
@@ -123,7 +123,7 @@
 			region.Record(p.stats.Batches.Observe(0))
 			region.Record(p.stats.Bytes.Observe(0))
 			region.Record(p.stats.Rows.Observe(0))
-			level.Error(p.logger).Log("msg", "caching pass-though, won't cache result")
+			level.Debug(p.logger).Log("msg", "caching pass-though, won't cache result")
 			return nil, err
 		}
 

diff --git a/pkg/engine/internal/planner/physical/cache.go b/pkg/engine/internal/planner/physical/cache.go
--- a/pkg/engine/internal/planner/physical/cache.go
+++ b/pkg/engine/internal/planner/physical/cache.go
@@ -115,8 +115,18 @@
 		}
 	} else {
 		// This is the expected code path:
-		// Inject Cache between the DataObjScan and its parent.
-		plan.graph.Inject(parents[0], cacheNode)
+		// Insert Cache between the DataObjScan and its parent.
+		parent := parents[0]
+		plan.graph.Add(cacheNode)
+		if err := plan.graph.RemoveEdge(dag.Edge[Node]{Parent: parent, Child: scan}); err != nil {
+			return err
+		}
+		if err := plan.graph.AddEdge(dag.Edge[Node]{Parent: parent, Child: cacheNode}); err != nil {
+			return err
+		}
+		if err := plan.graph.AddEdge(dag.Edge[Node]{Parent: cacheNode, Child: scan}); err != nil {
+			return err
+		}
 	}
 
 	return nil

diff --git a/pkg/engine/internal/planner/physical/cache_test.go b/pkg/engine/internal/planner/physical/cache_test.go
--- a/pkg/engine/internal/planner/physical/cache_test.go
+++ b/pkg/engine/internal/planner/physical/cache_test.go
@@ -153,3 +153,34 @@
 		require.Equal(t, node, root)
 	})
 }
+
+func TestWrapDataObjScansWithCache(t *testing.T) {
+	t.Run("inserts cache only on DataObjScan edge", func(t *testing.T) {
+		var g dag.Graph[Node]
+		parent := g.Add(&Batching{BatchSize: 100})
+		scan := g.Add(&DataObjScan{Location: "loc1"})
+		sibling := g.Add(&Filter{})
+		require.NoError(t, g.AddEdge(dag.Edge[Node]{Parent: parent, Child: scan}))
+		require.NoError(t, g.AddEdge(dag.Edge[Node]{Parent: parent, Child: sibling}))
+
+		plan := FromGraph(g)
+		require.NoError(t, WrapDataObjScansWithCache(context.Background(), "tenant1", plan, 0))
+
+		children := plan.graph.Children(parent)
+		require.Len(t, children, 2)
+		require.Contains(t, children, sibling)
+
+		var inserted *Cache
+		for _, child := range children {
+			if c, ok := child.(*Cache); ok {
+				inserted = c
+				break
+			}
+		}
+		require.NotNil(t, inserted, "expected a Cache node to be inserted")
+		require.Equal(t, TaskCacheDataObjScanResult, inserted.CacheName)
+		require.Equal(t, []Node{parent}, plan.graph.Parents(inserted))
+		require.Equal(t, []Node{scan}, plan.graph.Children(inserted))
+		require.Equal(t, []Node{inserted}, plan.graph.Parents(scan))
+	})
+}

diff --git a/pkg/engine/internal/util/dag/dag.go b/pkg/engine/internal/util/dag/dag.go
--- a/pkg/engine/internal/util/dag/dag.go
+++ b/pkg/engine/internal/util/dag/dag.go
@@ -103,6 +103,32 @@
 	return nil
 }
 
+// RemoveEdge removes a directed edge between two existing nodes in the graph.
+//
+// RemoveEdge returns an error if:
+//
+// * Either node is the zero value, or
+// * either node doesn't exist in the graph, or
+// * the edge doesn't exist.
+func (g *Graph[NodeType]) RemoveEdge(e Edge[NodeType]) error {
+	if isZero(e.Parent) || isZero(e.Child) {
+		return fmt.Errorf("parent and child nodes must not be zero values")
+	}
+	if !g.nodes.Contains(e.Parent) {
+		return fmt.Errorf("parent node %s does not exist in graph", e.Parent.ID())
+	}
+	if !g.nodes.Contains(e.Child) {
+		return fmt.Errorf("child node %s does not exist in graph", e.Child.ID())
+	}
+	if !slices.Contains(g.children[e.Parent], e.Child) {
+		return fmt.Errorf("edge %s -> %s does not exist in graph", e.Parent.ID(), e.Child.ID())
+	}
+
+	g.children[e.Parent] = slices.DeleteFunc(g.children[e.Parent], func(check NodeType) bool { return check == e.Child })
+	g.parents[e.Child] = slices.DeleteFunc(g.parents[e.Child], func(check NodeType) bool { return check == e.Parent })
+	return nil
+}
+
 // Eliminate removes the node n from the graph and reconnects n's parents to
 // n's children, maintaining connectivity across the graph.
 //

diff --git a/pkg/engine/internal/util/dag/dag_test.go b/pkg/engine/internal/util/dag/dag_test.go
--- a/pkg/engine/internal/util/dag/dag_test.go
+++ b/pkg/engine/internal/util/dag/dag_test.go
@@ -36,6 +36,17 @@
 		require.ErrorContains(t, err, "does not exist in graph")
 	})
 
+	t.Run("removing an existing edge updates parent and child", func(t *testing.T) {
+		var g dag.Graph[*testNode]
+		parent := g.Add(&testNode{id: ulid.Make()})
+		child := g.Add(&testNode{id: ulid.Make()})
+		require.NoError(t, g.AddEdge(dag.Edge[*testNode]{Parent: parent, Child: child}))
+
+		require.NoError(t, g.RemoveEdge(dag.Edge[*testNode]{Parent: parent, Child: child}))
+		require.Empty(t, g.Children(parent))
+		require.Empty(t, g.Parents(child))
+	})
+
 	t.Run("adding an edge for zero value nodes fails", func(t *testing.T) {
 		var g dag.Graph[*testNode]

_{You can send follow-ups to this agent here.}

[cursor]

Inject moves all parent children, not just DataObjScan

Low Severity

plan.graph.Inject(parents[0], cacheNode) moves ALL children of parents[0] under cacheNode, not just the scan node. The Inject method replaces the entire children list of the parent. If parents[0] ever has additional children besides the DataObjScan, those children would incorrectly end up under the cache node, corrupting the plan graph.