remove scans batching

Author: salvacorts

docs/sources/shared/configuration.md

@@ -4937,11 +4937,6 @@ otlp_config:
 # CLI flag: -limits.debug-engine-streams
 [debug_engine_streams: <boolean> | default = false]
 
-# Experimental: Number of PointersScan targets to batch per task at the workflow
-# level. 0 = one task per path.
-# CLI flag: -limits.metastore-pointers-scans-per-task
-[metastore_pointers_scans_per_task: <int> | default = 0]
-
 # Experimental: Maximum size of record batches sent to sinks when draining the
 # pipeline (e.g. 1MB). 0 = no batching, send each record as-is.
 # CLI flag: -limits.record-batch-size

pkg/engine/engine.go

@@ -367,7 +367,7 @@ func (e *Engine) buildPhysicalPlan(ctx context.Context, tenantID string, logger
 }
 
 func (e *Engine) metastoreSectionsResolver(ctx context.Context, tenantID string) physical.MetastoreSectionsResolver {
-	planner := physical.NewMetastorePlanner(e.metastore, e.limits.MetastorePointersScansPerTask(tenantID))
+	planner := physical.NewMetastorePlanner(e.metastore)
 	return func(selector physical.Expression, predicates []physical.Expression, start time.Time, end time.Time) ([]*metastore.DataobjSectionDescriptor, error) {
 		ctx, region := xcap.StartRegion(ctx, "ObjectMetastore.Sections")
 		defer region.End()

pkg/engine/internal/planner/physical/metastore_planner.go

@@ -13,17 +13,11 @@ import (
 
 type MetastorePlanner struct {
 	metastore metastore.Metastore
-
-	// PointersScansPerTask is the batch size for workflow-level sharding: when > 0,
-	// the ScanSet gets ShardBatchSize set so each task runs up to this many PointersScan targets.
-	// 0 means one task per path (current behavior).
-	PointersScansPerTask int
 }
 
-func NewMetastorePlanner(metastore metastore.Metastore, pointersScansPerTask int) MetastorePlanner {
+func NewMetastorePlanner(metastore metastore.Metastore) MetastorePlanner {
 	return MetastorePlanner{
-		metastore:            metastore,
-		PointersScansPerTask: pointersScansPerTask,
+		metastore: metastore,
 	}
 }
 
@@ -51,8 +45,7 @@ func (p MetastorePlanner) Plan(ctx context.Context, selector Expression, predica
 	plan.graph.Add(parallelize)
 
 	scanSet := &ScanSet{
-		NodeID:         ulid.Make(),
-		ShardBatchSize: p.PointersScansPerTask,
+		NodeID: ulid.Make(),
 	}
 	plan.graph.Add(scanSet)
 

pkg/engine/internal/planner/physical/metastore_planner_test.go

@@ -38,8 +38,7 @@ func TestMetastorePlanner_Plan_UsesMergeRootAndPointersTargets(t *testing.T) {
 	start := time.Unix(10, 0)
 	end := start.Add(time.Hour)
 
-	const shardSize = 2
-	p := NewMetastorePlanner(ms, shardSize)
+	p := NewMetastorePlanner(ms)
 	plan, err := p.Plan(context.Background(), nil, nil, start, end)
 	require.NoError(t, err)
 
@@ -66,6 +65,4 @@ func TestMetastorePlanner_Plan_UsesMergeRootAndPointersTargets(t *testing.T) {
 		require.Equal(t, start, target.Pointers.Start)
 		require.Equal(t, end, target.Pointers.End)
 	}
-
-	require.Equal(t, shardSize, set.ShardBatchSize)
 }

pkg/engine/internal/planner/physical/scanset.go

@@ -71,11 +71,6 @@ type ScanSet struct {
 	// returned. Predicates would almost always contain a time range filter to
 	// only read the logs for the requested time range.
 	Predicates []Expression
-
-	// ShardBatchSize, when > 0, causes Shards() to yield ScanSet nodes with up to
-	// ShardBatchSize targets each (workflow-level batching). Applies to all target types.
-	// 0 means one shard per target.
-	ShardBatchSize int
 }
 
 // ID returns the ULID that uniquely identifies the node in the plan.
@@ -89,9 +84,8 @@ func (s *ScanSet) Clone() Node {
 	}
 
 	return &ScanSet{
-		NodeID:         ulid.Make(),
-		Targets:        newTargets,
-		ShardBatchSize: s.ShardBatchSize,
+		NodeID:  ulid.Make(),
+		Targets: newTargets,
 	}
 }
 
@@ -104,40 +98,9 @@ func (s *ScanSet) Type() NodeType {
 // will be a clone. Projections and predicates on the ScanSet are cloned and
 // applied to each shard.
 //
-// When ShardBatchSize > 0, yields ScanSet nodes each with up to ShardBatchSize
-// targets (workflow-level batching). Otherwise yields one node per target
-// (DataObjScan or PointersScan).
-//
 // Shards panics if one of the targets is invalid.
 func (s *ScanSet) Shards() iter.Seq[Node] {
 	return func(yield func(Node) bool) {
-		// Batched sharding: when ShardBatchSize > 0, yield one ScanSet per chunk of targets.
-		if s.ShardBatchSize > 0 {
-			for i := 0; i < len(s.Targets); i += s.ShardBatchSize {
-				end := i + s.ShardBatchSize
-				if end > len(s.Targets) {
-					end = len(s.Targets)
-				}
-				chunk := s.Targets[i:end]
-				batchedTargets := make([]*ScanTarget, len(chunk))
-				for j, t := range chunk {
-					batchedTargets[j] = t.Clone()
-				}
-				shardSet := &ScanSet{
-					NodeID:         ulid.Make(),
-					Targets:        batchedTargets,
-					Projections:    s.Projections,
-					Predicates:     s.Predicates,
-					ShardBatchSize: 0, // injected ScanSet does not re-shard
-				}
-				if !yield(shardSet) {
-					return
-				}
-			}
-			return
-		}
-
-		// One shard per target (current behavior).
 		for _, target := range s.Targets {
 			switch target.Type {
 			case ScanTypeDataObject:

pkg/engine/internal/planner/physical/scanset_test.go

@@ -2,7 +2,6 @@ package workflow
 
 import (
 	"context"
-	"fmt"
 	"testing"
 	"time"
 
@@ -29,7 +28,7 @@ func TestPlanWorkflow_MetastorePlan_UsesMergeRootAndPointersPartitions(t *testin
 	start := time.Unix(10, 0)
 	end := start.Add(time.Hour)
 
-	p := physical.NewMetastorePlanner(ms, 0)
+	p := physical.NewMetastorePlanner(ms)
 	plan, err := p.Plan(context.Background(), nil, nil, start, end)
 	require.NoError(t, err)
 
@@ -52,7 +51,7 @@ func TestPlanWorkflow_MetastorePlan_UsesMergeRootAndPointersPartitions(t *testin
 	for _, child := range children {
 		childRoot, err := child.Fragment.Root()
 		require.NoError(t, err)
-		require.IsType(t, &physical.PointersScan{}, childRoot, "partition fragment root should be PointersScan when not batching")
+		require.IsType(t, &physical.PointersScan{}, childRoot)
 		gotLocations[childRoot.(*physical.PointersScan).Location] = struct{}{}
 	}
 
@@ -61,40 +60,3 @@ func TestPlanWorkflow_MetastorePlan_UsesMergeRootAndPointersPartitions(t *testin
 		require.True(t, ok, "missing partition for %q", indexPath)
 	}
 }
-
-func TestPlanWorkflow_MetastorePlan_BatchesPointersScans(t *testing.T) {
-	paths := make([]string, 250)
-	for i := range paths {
-		paths[i] = fmt.Sprintf("index/%d", i)
-	}
-	ms := fakeMetastoreIndexes{indexPaths: paths}
-	start := time.Unix(10, 0)
-	end := start.Add(time.Hour)
-
-	p := physical.NewMetastorePlanner(ms, 100)
-	plan, err := p.Plan(context.Background(), nil, nil, start, end)
-	require.NoError(t, err)
-
-	graph, err := planWorkflow("tenant", plan)
-	require.NoError(t, err)
-
-	rootTask, err := graph.Root()
-	require.NoError(t, err)
-
-	rootNode, err := rootTask.Fragment.Root()
-	require.NoError(t, err)
-	require.IsType(t, &physical.Merge{}, rootNode)
-
-	children := graph.Children(rootTask)
-	require.Len(t, children, 3, "250 paths with batch 100 should produce 3 tasks (100, 100, 50)")
-
-	expectedSizes := []int{100, 100, 50}
-	for i, child := range children {
-		childRoot, err := child.Fragment.Root()
-		require.NoError(t, err)
-		scanSet, ok := childRoot.(*physical.ScanSet)
-		require.True(t, ok, "child %d: partition fragment root should be ScanSet when batching, got %T", i, childRoot)
-		require.Len(t, scanSet.Targets, expectedSizes[i], "child %d: expected %d PointersScan targets", i, expectedSizes[i])
-		require.Equal(t, 0, scanSet.ShardBatchSize, "batched shard should have ShardBatchSize 0")
-	}
-}

pkg/engine/internal/workflow/metastore_workflow_planner_test.go

@@ -266,10 +266,6 @@ func (l *limiter) DebugEngineStreams(_ string) bool {
 	return false // This setting for the v2 execution engine is unused in LogCLI
 }
 
-func (l *limiter) MetastorePointersScansPerTask(_ string) int {
-	return 0 // This setting for the v2 execution engine is unused in LogCLI
-}
-
 func (l *limiter) RecordBatchSize(_ string) int {
 	return 0 // This setting for the v2 execution engine is unused in LogCLI
 }

pkg/logcli/client/file.go

@@ -27,7 +27,6 @@ type Limits interface {
 	MaxScanTaskParallelism(string) int
 	DebugEngineTasks(string) bool
 	DebugEngineStreams(string) bool
-	MetastorePointersScansPerTask(string) int
 	RecordBatchSize(string) int
 }
 
@@ -40,11 +39,10 @@ type fakeLimits struct {
 	multiVariantQueryEnable bool
 
 	// v2 engine limits
-	maxScanTaskParallelism        int
-	debugEngineTasks              bool
-	debugEngineStreams            bool
-	metastorePointersScansPerTask int
-	recordBatchSize               int
+	maxScanTaskParallelism int
+	debugEngineTasks       bool
+	debugEngineStreams     bool
+	recordBatchSize        int
 }
 
 func (f fakeLimits) MaxQuerySeries(_ context.Context, _ string) int {
@@ -83,10 +81,6 @@ func (f fakeLimits) DebugEngineStreams(_ string) bool {
 	return f.debugEngineStreams
 }
 
-func (f fakeLimits) MetastorePointersScansPerTask(_ string) int {
-	return f.metastorePointersScansPerTask
-}
-
 func (f fakeLimits) RecordBatchSize(_ string) int {
 	return f.recordBatchSize
 }

pkg/logql/limits.go

@@ -1549,10 +1549,6 @@ func (f fakeLimits) DebugEngineStreams(_ string) bool {
 	return false
 }
 
-func (f fakeLimits) MetastorePointersScansPerTask(_ string) int {
-	return 0
-}
-
 func (f fakeLimits) RecordBatchSize(_ string) int {
 	return 0
 }