feat(log): add time clipping push down (openGemini#584)

Signed-off-by: noFloat <[email protected]>

Author: noFloat

engine/comm/cursor.go

@@ -42,6 +42,11 @@ type KeyCursor interface {
 	NextAggData() (*record.Record, *FileInfo, error)
 }
 
+type TimeCutKeyCursor interface {
+	KeyCursor
+	UpdateTime(time int64)
+}
+
 type FileInfo struct {
 	MinTime    int64
 	MaxTime    int64

engine/hybrid_store_reader.go

@@ -247,7 +247,7 @@ func (r *HybridStoreReader) initDetachedFileReader(frags executor.IndexFrags) (c
 	if err != nil {
 		return nil, err
 	}
-	if r.schema.Options().CanTimeLimitPushDown() {
+	if r.schema.Options().CanLimitPushDown() {
 		sortLimitCursor := immutable.NewSortLimitCursor(r.schema.Options(), r.readerCtx.GetSchemas(), fileReader)
 		return sortLimitCursor, nil
 	}

engine/hybridqp/compiler.go

@@ -60,6 +60,7 @@ type Options interface {
 	GetIterId() int32
 	IsIncQuery() bool
 	CanTimeLimitPushDown() bool
+	CanLimitPushDown() bool
 	SetPromQuery(bool)
 	IsPromQuery() bool
 	IsPromInstantQuery() bool

engine/immutable/detached_metadata.go

@@ -61,7 +61,6 @@ type SegmentTask struct {
 func (reader *DetachedMetaDataReader) InitReadBatch(s []*SegmentMeta, schema record.Schemas) {
 	reader.task = make(map[int]*SegmentTask, len(s))
 	reader.positionMap = make(map[int64]*position, len(s))
-
 	offset := make([]int64, 0, len(s)*len(schema))
 	length := make([]int64, 0, len(s)*len(schema))
 	startSegmentIndex := -1

engine/immutable/sort_limit_cursor.go

@@ -29,11 +29,11 @@ type SortLimitCursor struct {
 	isRead   bool
 	options  hybridqp.Options
 	schemas  record.Schemas
-	input    comm.KeyCursor
+	input    comm.TimeCutKeyCursor
 	sortHeap *SortLimitRows
 }
 
-func NewSortLimitCursor(options hybridqp.Options, schemas record.Schemas, input comm.KeyCursor) *SortLimitCursor {
+func NewSortLimitCursor(options hybridqp.Options, schemas record.Schemas, input comm.TimeCutKeyCursor) *SortLimitCursor {
 	sortIndex := make([]int, len(options.GetSortFields()))
 	for fk, field := range options.GetSortFields() {
 		for sk, v := range schemas {
@@ -116,6 +116,10 @@ func (t *SortLimitCursor) Next() (*record.Record, comm.SeriesInfoIntf, error) {
 				heap.Push(t.sortHeap, data)
 			}
 		}
+
+		if t.options.CanTimeLimitPushDown() && t.sortHeap.Len() >= t.options.GetLimit() {
+			t.input.UpdateTime(t.sortHeap.rows[0][len(t.schemas)-1].(int64))
+		}
 	}
 	t.isRead = true
 	return t.sortHeap.PopToRec(), nil, nil

engine/immutable/sort_limit_cursor_test.go

@@ -253,3 +253,6 @@ func (t *MockCursor) SetOps(ops []*comm.CallOption) {
 func (t *MockCursor) NextAggData() (*record.Record, *comm.FileInfo, error) {
 	return nil, nil, nil
 }
+
+func (t *MockCursor) UpdateTime(time int64) {
+}

engine/immutable/tssp_file_detached_reader.go

@@ -50,6 +50,7 @@ type TSSPFileDetachedReader struct {
 	filterSeqTime   int64
 	filterSeqId     int64
 	filterSeqFunc   func(int64, int64) bool
+	tr              util.TimeRange
 
 	metaDataQueue   MetaControl
 	metaIndex       []*MetaIndex
@@ -74,9 +75,10 @@ func NewTSSPFileDetachedReader(metaIndex []*MetaIndex, blocks [][]int, ctx *File
 		blocks:        blocks,
 		obsOpts:       path.Option(),
 		ctx:           ctx,
-		metaDataQueue: NewMetaControl(true, chunkMetaReadNum),
+		metaDataQueue: NewMetaControl(ctx.readCtx.Ascending, chunkMetaReadNum),
 		unnest:        unnest,
 		seqIndex:      -1,
+		tr:            util.TimeRange{Min: ctx.tr.Min, Max: ctx.tr.Max},
 	}
 	if config.IsLogKeeper() && options.GetLogQueryCurrId() != "" && options.GetLimit() > 0 {
 		err := r.parseSeqId(options)
@@ -203,6 +205,14 @@ func (t *TSSPFileDetachedReader) GetSchema() record.Schemas {
 	return t.ctx.schemas
 }
 
+func (t *TSSPFileDetachedReader) UpdateTime(time int64) {
+	if t.ctx.readCtx.Ascending {
+		t.tr.Max = time
+	} else {
+		t.tr.Min = time
+	}
+}
+
 func (t *TSSPFileDetachedReader) Next() (*record.Record, comm.SeriesInfoIntf, error) {
 	for {
 		if !t.isInit {
@@ -251,9 +261,9 @@ func (t *TSSPFileDetachedReader) readBatch() (*record.Record, error) {
 func (t *TSSPFileDetachedReader) filterData(rec *record.Record) *record.Record {
 	if rec != nil && (t.ctx.isOrder || t.isSort) {
 		if t.ctx.readCtx.Ascending {
-			rec = FilterByTime(rec, t.ctx.tr)
+			rec = FilterByTime(rec, t.tr)
 		} else {
-			rec = FilterByTimeDescend(rec, t.ctx.tr)
+			rec = FilterByTimeDescend(rec, t.tr)
 		}
 	}
 
@@ -304,7 +314,10 @@ func (t *TSSPFileDetachedReader) initChunkMeta() (bool, error) {
 	chunkMetas := make([]*SegmentMeta, 0)
 	for len(chunkMetas) < chunkReadNum && !t.metaDataQueue.IsEmpty() {
 		s, _ := t.metaDataQueue.Pop()
-		chunkMetas = append(chunkMetas, s.(*SegmentMeta))
+		currTr := s.(*SegmentMeta).chunkMeta.GetTimeRangeBy(s.(*SegmentMeta).id)
+		if currTr[0] <= t.tr.Max && currTr[1] >= t.tr.Min {
+			chunkMetas = append(chunkMetas, s.(*SegmentMeta))
+		}
 	}
 	t.dataReader.InitReadBatch(chunkMetas, t.ctx.schemas)
 	return true, nil

engine/immutable/tssp_file_detached_reader_test.go

@@ -419,3 +419,52 @@ func TestSeqFilterReader(t *testing.T) {
 
 	config.SetProductType("basic")
 }
+
+func TestSeqFilterReaderWhenUpdataBy(t *testing.T) {
+	testCompDir := t.TempDir()
+	config.SetProductType("logkeeper")
+	_ = fileops.RemoveAll(testCompDir)
+	sig := interruptsignal.NewInterruptSignal()
+	defer func() {
+		sig.Close()
+		_ = fileops.RemoveAll(testCompDir)
+	}()
+	mstName := "mst"
+	err := writeData(testCompDir, mstName)
+	if err != nil {
+		t.Errorf(err.Error())
+	}
+
+	p := path.Join(testCompDir, mstName)
+	reader, _ := NewDetachedMetaIndexReader(p, nil)
+
+	metaIndex, _ := reader.ReadMetaIndex([]int64{16, 56}, []int64{40, 40})
+	option := query2.ProcessorOptions{
+		QueryId: 1,
+		Query:   "select * from mst",
+	}
+	option.LogQueryCurrId = "1635724829000000000|9^^"
+	option.Limit = 10
+	schema3 := []record.Field{
+		{Name: record.SeqIDField, Type: influx.Field_Type_Int},
+		{Name: "field2_int", Type: influx.Field_Type_Int},
+		{Name: "time", Type: influx.Field_Type_Int},
+	}
+	decs := NewFileReaderContext(util.TimeRange{Min: 1635724829000000000, Max: 1645724819000000000}, schema3, NewReadContext(true), NewFilterOpts(nil, nil, nil, nil), nil, false)
+	treader, _ := NewTSSPFileDetachedReader(metaIndex[:1], [][]int{[]int{0, 2}}, decs, sparseindex.NewOBSFilterPath("", p, nil), nil, true, &option)
+	totalRow := 0
+	treader.UpdateTime(1635724839000000000)
+	for {
+		data, _, _ := treader.Next()
+		if data == nil {
+			break
+		}
+		assert.Equal(t, data.RowNums(), data.ColVals[0].Len)
+		totalRow += data.RowNums()
+	}
+	if totalRow != 0 {
+		t.Errorf("tssp_file_detached update time wrong")
+	}
+
+	config.SetProductType("basic")
+}

lib/util/lifted/influx/query/select.go

@@ -579,10 +579,14 @@ func (opt *ProcessorOptions) GetOptDimension() []string {
 	return opt.Dimensions
 }
 
-func (opt *ProcessorOptions) CanTimeLimitPushDown() bool {
+func (opt *ProcessorOptions) CanLimitPushDown() bool {
 	return opt.Limit > 0 && len(opt.SortFields) > 0
 }
 
+func (opt *ProcessorOptions) CanTimeLimitPushDown() bool {
+	return opt.CanLimitPushDown() && opt.SortFields[0].Name == "time"
+}
+
 // Zone returns the zone information for the given time. The offset is in nanoseconds.
 func (opt *ProcessorOptions) Zone(ns int64) (string, int64) {
 	if opt.Location == nil {