Coverage Report

Created: 2026-11-05 13:02

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/io/fs/stream_sink_file_writer.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <brpc/stream.h>
21
#include <gen_cpp/olap_common.pb.h>
22
#include <gen_cpp/types.pb.h>
23
24
#include <queue>
25
#include <unordered_set>
26
27
#include "io/fs/file_writer.h"
28
#include "util/uid_util.h"
29
30
namespace doris {
31
32
class LoadStreamStub;
33
34
struct RowsetId;
35
struct SegmentStatistics;
36
37
namespace io {
38
struct FileCacheAllocatorBuilder;
39
class StreamSinkFileWriter final : public FileWriter {
40
public:
41
    StreamSinkFileWriter(std::vector<std::shared_ptr<LoadStreamStub>> streams)
42
7
            : _streams(std::move(streams)) {}
43
44
    void init(PUniqueId load_id, int64_t partition_id, int64_t index_id, int64_t tablet_id,
45
              int32_t segment_id, FileType file_type = FileType::SEGMENT_FILE);
46
47
    Status appendv(const Slice* data, size_t data_cnt) override;
48
49
0
    size_t bytes_appended() const override { return _bytes_appended; }
50
51
10
    State state() const override { return _state; }
52
53
    // FIXME(plat1ko): Maybe it's an inappropriate abstraction?
54
0
    const Path& path() const override {
55
0
        static Path dummy;
56
0
        return dummy;
57
0
    }
58
    Status close(bool non_block = false) override;
59
60
private:
61
    Status _finalize();
62
    std::unordered_set<int64_t> _get_fault_injection_failed_dst_ids() const;
63
    std::vector<std::shared_ptr<LoadStreamStub>> _streams;
64
65
    PUniqueId _load_id;
66
    int64_t _partition_id;
67
    int64_t _index_id;
68
    int64_t _tablet_id;
69
    int32_t _segment_id;
70
    size_t _bytes_appended = 0;
71
    State _state {State::OPENED};
72
    FileType _file_type {FileType::SEGMENT_FILE};
73
};
74
75
} // namespace io
76
} // namespace doris