ydb/core/tx/columnshard/columnshard__index_scan.h


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

#pragma once

#include "columnshard__scan.h"
#include "columnshard_common.h"
#include <ydb/core/tx/columnshard/engines/indexed_read_data.h>

namespace NKikimr::NColumnShard {

class TIndexColumnResolver : public IColumnResolver {
    const NOlap::TIndexInfo& IndexInfo;

public:
    explicit TIndexColumnResolver(const NOlap::TIndexInfo& indexInfo)
        : IndexInfo(indexInfo)
    {}

    TString GetColumnName(ui32 id, bool required) const override {
        return IndexInfo.GetColumnName(id, required);
    }

    const NTable::TScheme::TTableSchema& GetSchema() const override {
        return IndexInfo;
    }
};


using NOlap::TUnifiedBlobId;
using NOlap::TBlobRange;

class TColumnShardScanIterator : public TScanIteratorBase {
private:
    NOlap::TReadMetadata::TConstPtr ReadMetadata;
    NOlap::TFetchBlobsQueue FetchBlobsQueue;
    NOlap::TIndexedReadData IndexedData;
    std::unordered_map<NOlap::TCommittedBlob, ui32, THash<NOlap::TCommittedBlob>> WaitCommitted;
    TDeque<NOlap::TPartialReadResult> ReadyResults;
    ui64 ItemsRead = 0;
    const i64 MaxRowsInBatch = 5000;
    NColumnShard::TDataTasksProcessorContainer DataTasksProcessor;
    NColumnShard::TScanCounters ScanCounters;
public:
    TColumnShardScanIterator(NOlap::TReadMetadata::TConstPtr readMetadata, NColumnShard::TDataTasksProcessorContainer processor, const NColumnShard::TScanCounters& scanCounters);
    ~TColumnShardScanIterator();

    virtual void Apply(IDataTasksProcessor::ITask::TPtr task) override;

    virtual bool HasWaitingTasks() const override;

    void AddData(const TBlobRange& blobRange, TString data) override;

    bool Finished() const  override {
        return FetchBlobsQueue.IsStopped() && ReadyResults.empty();
    }

    NOlap::TPartialReadResult GetBatch() override;

    TBlobRange GetNextBlobToRead() override;

    size_t ReadyResultsCount() const override {
        return ReadyResults.size();
    }

private:
    void FillReadyResults();
};

}