(done bool, n int)
| 75 | } |
| 76 | |
| 77 | func (v *VectorReader) ConcurrentPull(done bool, n int) (vector.Any, error) { |
| 78 | if done { |
| 79 | return nil, v.close() |
| 80 | } |
| 81 | if err := v.ctx.Err(); err != nil { |
| 82 | v.close() |
| 83 | return nil, err |
| 84 | } |
| 85 | for { |
| 86 | if n := len(v.vecs); n > 0 { |
| 87 | // Return these last to first so v.vecs gets resued. |
| 88 | vec := v.vecs[n-1] |
| 89 | v.vecs = v.vecs[:n-1] |
| 90 | return vec, nil |
| 91 | } |
| 92 | hdr, off, err := v.stream.next() |
| 93 | if hdr == nil || err != nil { |
| 94 | return nil, err |
| 95 | } |
| 96 | o, err := csup.NewObjectFromHeader(io.NewSectionReader(v.readerAt, off, math.MaxInt64), *hdr) |
| 97 | if err != nil { |
| 98 | return nil, err |
| 99 | } |
| 100 | // XXX using the query context for the metadata filter unnecessarily |
| 101 | // pollutes the type context. We should use the csup local context for |
| 102 | // this filtering but this will require a little compiler refactoring to be |
| 103 | // able to build runtime expressions that use different type contexts. |
| 104 | if len(v.metaFilters) == 0 || !pruneObject(v.sctx, v.metaFilters[n], o) { |
| 105 | vo := vcache.NewObjectFromCSUP(o) |
| 106 | if v.pushdown.Unordered() { |
| 107 | v.vecs, err = vo.FetchUnordered(v.vecs[:0], v.sctx, v.pushdown.Projection()) |
| 108 | if err != nil { |
| 109 | return nil, err |
| 110 | } |
| 111 | } else { |
| 112 | vec, err := vo.Fetch(v.sctx, v.pushdown.Projection()) |
| 113 | if err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | v.vecs = append(v.vecs, vec) |
| 117 | } |
| 118 | } |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | func pruneObject(sctx *super.Context, mf *metafilter, o *csup.Object) bool { |
| 123 | vals := o.ProjectMetadata(sctx, mf.projection) |
no test coverage detected