chore(async search): search per N-minute intervals

asyncsearcher/async_searcher_test.go

@@ -40,16 +40,7 @@ type fakeDP struct {
 
 type fakeFractionProvider fracmanager.List
 
-func (fp fakeFractionProvider) AcquireFraction(name string) (frac.Fraction, func(), bool) {
-	for _, f := range fp {
-		if f.Info().Name() == name {
-			return f, func() {}, true
-		}
-	}
-	return nil, func() {}, false
-}
-
-func (fp fakeFractionProvider) AcquireFractions() (fracmanager.List, func()) {
+func (fp fakeFractionProvider) AcquireFractionsInRange(from, to seq.MID) (fracmanager.List, func()) {
 	return fracmanager.List(fp), func() {}
 }
 
@@ -79,34 +70,9 @@ func TestAsyncSearcherMaintain(t *testing.T) {
 	as.processWg.Wait()
 }
 
-// partialProvider lists both fractions (so info.Fractions gets two entries),
-// but only "present" is acquirable. "missing" simulates a fraction that was
-// removed by the time the search reached it: it stays listed in info.Fractions
-// yet never produces a .qpr file.
-type partialProvider struct {
-	list fracmanager.List
-}
-
-func (p *partialProvider) AcquireFractions() (fracmanager.List, func()) {
-	return p.list, func() {}
-}
-
-func (p *partialProvider) AcquireFraction(name string) (frac.Fraction, func(), bool) {
-	for _, f := range p.list {
-		if f.Info().Name() == name && name == "present" {
-			return f, func() {}, true
-		}
-	}
-	return nil, func() {}, false
-}
-
-// TestMergeSkipsMissingFrac is a regression test: when a fraction listed in
-// info.Fractions was skipped (already removed) and produced no .qpr, merge used
-// to build its path from info.Fractions, hit a missing file in loadSearchResult,
-// discard the whole accumulated result, write an empty .mqpr and delete the
-// real .qpr — losing the only matching document.
-func TestMergeSkipsMissingFrac(t *testing.T) {
+func TestMerge(t *testing.T) {
 	r := require.New(t)
+	now := time.Now()
 
 	cfg := AsyncSearcherConfig{DataDir: t.TempDir()}
 	mp, err := mappingprovider.New("", mappingprovider.WithMapping(seq.Mapping{}))
@@ -115,29 +81,104 @@ func TestMergeSkipsMissingFrac(t *testing.T) {
 	as := MustStartAsync(cfg, mp, nil)
 	t.Cleanup(func() { as.readOnly.Store(false) })
 
-	presentFrac := &fakeFrac{
-		info: common.Info{Path: "present"},
-		dp:   fakeDP{qpr: seq.QPR{IDs: []seq.IDSource{{ID: seq.ID{MID: 42}}}, Total: 1}},
+	frac1 := &fakeFrac{
+		info: common.Info{Path: "1", From: seq.TimeToMID(now.Add(-time.Minute * 11)), To: seq.TimeToMID(now.Add(-time.Minute * 6))},
+		dp:   fakeDP{qpr: seq.QPR{IDs: []seq.IDSource{{ID: seq.ID{MID: 1}}}, Total: 1}},
+	}
+	frac2 := &fakeFrac{
+		info: common.Info{Path: "2", From: seq.TimeToMID(now.Add(-time.Minute * 6)), To: seq.TimeToMID(now.Add(-time.Minute * 1))},
+		dp:   fakeDP{qpr: seq.QPR{IDs: []seq.IDSource{{ID: seq.ID{MID: 2}}}, Total: 1}},
 	}
-	missingFrac := &fakeFrac{info: common.Info{Path: "missing"}}
-	provider := &partialProvider{list: fracmanager.List{presentFrac, missingFrac}}
+	provider := &fakeFractionProvider{frac1, frac2}
 
 	req := AsyncSearchRequest{
-		ID:        uuid.New().String(),
-		Params:    processor.SearchParams{Limit: 1000, Order: seq.DocsOrderDesc},
+		ID: uuid.New().String(),
+		Params: processor.SearchParams{
+			Limit: 1000,
+			Order: seq.DocsOrderDesc,
+			From:  seq.TimeToMID(now.UTC().Add(-time.Minute * 30).Truncate(time.Millisecond)),
+			To:    seq.TimeToMID(now.UTC().Truncate(time.Millisecond)),
+		},
 		Query:     "*",
 		Retention: time.Hour,
 	}
 	r.NoError(as.StartSearch(req, provider))
 	as.processWg.Wait()
 
-	// "missing" produced no .qpr; "present" did. Merge must not drop the
-	// present result while collapsing the request into a single .mqpr.
 	as.merge()
 
 	resp, ok := as.FetchSearchResult(FetchSearchResultRequest{ID: req.ID, Limit: 1000, Order: seq.DocsOrderDesc})
 	r.True(ok)
 	r.Equal(AsyncSearchStatusDone, resp.Status)
-	r.Len(resp.QPR.IDs, 1)
-	r.Equal(seq.MID(42), resp.QPR.IDs[0].ID.MID)
+	r.Len(resp.QPR.IDs, 2)
+	r.Equal(seq.MID(2), resp.QPR.IDs[0].ID.MID)
+	r.Equal(seq.MID(1), resp.QPR.IDs[1].ID.MID)
+}
+
+func TestBuildIntervals(t *testing.T) {
+	tests := []struct {
+		name     string
+		from     seq.MID
+		to       seq.MID
+		expected []searchInterval
+	}{
+		{
+			name:     "empty_range_from_equals_to",
+			from:     100,
+			to:       100,
+			expected: nil,
+		},
+		{
+			name: "single_interval_small_range",
+			from: 0,
+			to:   100,
+			expected: []searchInterval{
+				{0, 100},
+			},
+		},
+		{
+			name: "single_interval_exact_split",
+			from: 0,
+			to:   seq.DurationToMID(defaultSearchInterval),
+			expected: []searchInterval{
+				{0, 300_000_000_000},
+			},
+		},
+		{
+			name: "two_intervals",
+			from: 0,
+			to:   seq.DurationToMID(defaultSearchInterval) * 2,
+			expected: []searchInterval{
+				{0, 299_999_999_999},
+				{300_000_000_000, 600_000_000_000},
+			},
+		},
+		{
+			name: "three_intervals_with_remainder",
+			from: 0,
+			to:   seq.DurationToMID(defaultSearchInterval)*3 + 50,
+			expected: []searchInterval{
+				{0, 299_999_999_999},
+				{300_000_000_000, 599_999_999_999},
+				{600_000_000_000, 899_999_999_999},
+				{900_000_000_000, 900_000_000_050},
+			},
+		},
+		{
+			name: "minimal_range",
+			from: 5,
+			to:   6,
+			expected: []searchInterval{
+				{5, 6},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := require.New(t)
+			result := buildIntervals(tt.from, tt.to)
+			r.Equal(tt.expected, result)
+		})
+	}
 }

fracmanager/fetcher.go

@@ -138,6 +138,20 @@ func groupIDsByFraction(idsOrig seq.IDSources, fracsIn List) (List, [][]seq.ID)
 	fracsOut := fracsIn.FilterInRange(minMID, maxMID) // reduce candidate fractions
 	idsByFracs := make([][]seq.ID, 0, len(fracsOut))
 
+	fracsByName := make(map[string]struct{}, len(fracsOut))
+	for _, f := range fracsOut {
+		fracsByName[f.Info().Name()] = struct{}{}
+	}
+	for i, id := range ids {
+		if id.Hint == "" {
+			continue
+		}
+		if _, ok := fracsByName[id.Hint]; !ok {
+			// we need this to check ids with non-existing hints for all fractions
+			ids[i].Hint = ""
+		}
+	}
+
 	// stats
 	withHintsCnt := 0
 	hintMissesCnt := 0

fracmanager/fracmanager.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ozontech/seq-db/frac"
 	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/logger"
+	"github.com/ozontech/seq-db/seq"
 	"github.com/ozontech/seq-db/storage"
 	"github.com/ozontech/seq-db/storage/s3"
 	"github.com/ozontech/seq-db/util"
@@ -157,6 +158,10 @@ func (fm *FracManager) AcquireFractions() (List, func()) {
 	return fm.lc.registry.acquireAllFractions()
 }
 
+func (fm *FracManager) AcquireFractionsInRange(from, to seq.MID) (List, func()) {
+	return fm.lc.registry.acquireFractionsInRange(from, to)
+}
+
 func (fm *FracManager) Oldest() uint64 {
 	return fm.lc.registry.oldestTotal()
 }

fracmanager/fraction_registry.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/ozontech/seq-db/frac"
 	"github.com/ozontech/seq-db/logger"
+	"github.com/ozontech/seq-db/seq"
 	"github.com/ozontech/seq-db/util"
 )
 
@@ -100,6 +101,14 @@ func (r *fractionRegistry) acquireAllFractions() ([]frac.Fraction, func()) {
 	return r.snapshot.AcquireAll()
 }
 
+// acquireFractionsInRange returns a read-only subset of fractions within the range
+func (r *fractionRegistry) acquireFractionsInRange(from, to seq.MID) ([]frac.Fraction, func()) {
+	r.muSnapshot.RLock()
+	defer r.muSnapshot.RUnlock()
+
+	return r.snapshot.AcquireInRange(from, to)
+}
+
 // statistics returns current size statistics of the registry.
 func (r *fractionRegistry) statistics() registryStats {
 	r.mu.RLock()

fracmanager/fractions_snapshot.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/ozontech/seq-db/frac"
+	"github.com/ozontech/seq-db/seq"
 )
 
 // RefCounter provides reference counting capability.
@@ -17,7 +18,7 @@ type RefCounter interface {
 // with associated reference counters to keep them alive.
 type fractionsSnapshot struct {
 	counters    []RefCounter    // Reference counters to keep fractions alive
-	fractions   []frac.Fraction // The actual fractions in chronological order
+	fractions   []frac.Fraction // The actual fractions
 	names       map[string]int
 	oldestLocal uint64
 	oldestTotal uint64
@@ -77,6 +78,28 @@ func (fs *fractionsSnapshot) AcquireAll() ([]frac.Fraction, func()) {
 	}
 }
 
+func (fs *fractionsSnapshot) AcquireInRange(from, to seq.MID) ([]frac.Fraction, func()) {
+	fracs := make(List, 0)
+	counters := make([]RefCounter, 0)
+
+	for i := range len(fs.fractions) {
+		f := fs.fractions[i]
+		c := fs.counters[i]
+
+		if f.IsIntersecting(from, to) {
+			fracs = append(fracs, f)
+			c.Inc()
+			counters = append(counters, c)
+		}
+	}
+
+	return fracs, func() {
+		for _, c := range counters {
+			c.Dec()
+		}
+	}
+}
+
 func (fs *fractionsSnapshot) AcquireOne(name string) (frac.Fraction, func(), bool) {
 	i, ok := fs.names[name]
 	if !ok {

tests/integration_tests/integration_test.go

@@ -1579,6 +1579,7 @@ func (s *IntegrationTestSuite) TestTimeField() {
 func (s *IntegrationTestSuite) TestAsyncSearch() {
 	t := s.T()
 	r := require.New(t)
+	now := time.Now()
 
 	cfg := *s.Config
 	cfg.Mapping = map[string]seq.MappingTypes{
@@ -1618,8 +1619,8 @@ func (s *IntegrationTestSuite) TestAsyncSearch() {
 
 	startReq := search.AsyncRequest{
 		Query:     "* | fields ip, method, uri",
-		From:      time.UnixMilli(0).UTC(),
-		To:        time.Now().UTC().Add(time.Hour).Truncate(time.Millisecond),
+		From:      now.UTC().Truncate(time.Millisecond),
+		To:        now.UTC().Add(time.Minute).Truncate(time.Millisecond),
 		Retention: time.Minute * 5,
 		Aggregations: []search.AggQuery{
 			{

tests/integration_tests/single_test.go

@@ -255,7 +255,7 @@ func (s *SingleTestSuite) TestFetchHints() {
 				fetched = append(fetched, string(doc.Data))
 			}
 		}
-		s.Assert().Empty(fetched)
+		s.Require().Equal(docStrs, fetched) // we will check id with broken hint in each fraction
 	})
 }