| 111 | } |
| 112 | |
| 113 | func (p *VectorReader) ConcurrentPull(done bool, id int) (vector.Any, error) { |
| 114 | if done { |
| 115 | return nil, nil |
| 116 | } |
| 117 | for { |
| 118 | if err := p.ctx.Err(); err != nil { |
| 119 | return nil, err |
| 120 | } |
| 121 | if p.rrs[id] == nil { |
| 122 | pr := p.fr.ParquetReader() |
| 123 | rowGroup := int(p.nextRowGroup.Add(1) - 1) |
| 124 | if rowGroup >= pr.NumRowGroups() { |
| 125 | return nil, nil |
| 126 | } |
| 127 | if len(p.metadataFilters) > 0 { |
| 128 | rgMetadata := pr.MetaData().RowGroup(rowGroup) |
| 129 | val := buildMetadataValue(p.sctx, rgMetadata, p.metadataColIndexes, p.colIndexToField) |
| 130 | if !p.metadataFilters[id].Eval(val).Ptr().AsBool() { |
| 131 | continue |
| 132 | } |
| 133 | } |
| 134 | rr, err := p.fr.GetRecordReader(p.ctx, p.colIndexes, []int{rowGroup}) |
| 135 | if err != nil { |
| 136 | return nil, err |
| 137 | } |
| 138 | p.rrs[id] = rr |
| 139 | } |
| 140 | batch, err := p.rrs[id].Read() |
| 141 | if err != nil { |
| 142 | if err == io.EOF { |
| 143 | p.rrs[id] = nil |
| 144 | continue |
| 145 | } |
| 146 | return nil, err |
| 147 | } |
| 148 | return p.vbs[id].build(array.RecordToStructArray(batch), false) |
| 149 | } |
| 150 | } |
| 151 | |
| 152 | type vectorBuilder struct { |
| 153 | sctx *super.Context |