be/src/format_v2/table/lance_reader.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 <cstddef> |
21 | | #include <cstdint> |
22 | | #include <memory> |
23 | | #include <optional> |
24 | | #include <string> |
25 | | #include <vector> |
26 | | |
27 | | #include "common/status.h" |
28 | | #include "format_v2/lance/lance_record_batch_converter.h" |
29 | | #include "format_v2/table_reader.h" |
30 | | #include "runtime/runtime_profile.h" |
31 | | |
32 | | struct LanceBatch; |
33 | | struct LanceDataset; |
34 | | struct LanceFtsQueryContext; |
35 | | struct LanceScanner; |
36 | | |
37 | | namespace doris { |
38 | | class ShardedKVCache; |
39 | | } |
40 | | |
41 | | namespace doris::format::lance { |
42 | | |
43 | | // A FORMAT_LANCE table reader. Unlike file formats such as Parquet, a Lance split is not a |
44 | | // physical-file range. It either selects fragments from a fixed snapshot or scans the whole |
45 | | // latest snapshot, so the dataset is owned by this table reader and each split owns its scanner. |
46 | | class LanceTableReader final : public TableReader { |
47 | | public: |
48 | | ~LanceTableReader() override; |
49 | | |
50 | | // Fetch the schema of a Lance dataset without initializing the scan path. Version zero opens |
51 | | // the latest snapshot, which is used by backend-local TVF schema discovery. |
52 | | Status fetch_schema(const TFileRangeDesc& range, const TFileScanRangeParams& scan_params, |
53 | | std::vector<std::string>* column_names, |
54 | | std::vector<DataTypePtr>* column_types) const; |
55 | | |
56 | | Status init(TableReadOptions&& options) override; |
57 | | Status prepare_split(const SplitReadOptions& options) override; |
58 | | Status get_block(Block* block, bool* eos) override; |
59 | | // Fetch top-level projected columns by native Lance row IDs from one fixed dataset snapshot. |
60 | | // Input order and duplicates are preserved by lance-c. Missing rows are rejected because row |
61 | | // IDs produced by phase one must still exist in the same snapshot during materialization. |
62 | | Status read_by_row_ids(const TFileRangeDesc& range, const std::vector<uint64_t>& row_ids, |
63 | | Block* block); |
64 | | Status abort_split() override; |
65 | | Status close() override; |
66 | | |
67 | | private: |
68 | | struct DatasetKey { |
69 | | std::string uri; |
70 | | int64_t version = 0; |
71 | | std::vector<std::string> storage_options; |
72 | 24 | bool operator==(const DatasetKey&) const = default; |
73 | | }; |
74 | | |
75 | | Status _resolve_search_kind(); |
76 | | Status _validate_external_search_request() const; |
77 | | Status _ensure_dataset_open(const TFileRangeDesc& range, bool prepare_fts_context = true); |
78 | | Status _open_dataset(const DatasetKey& key); |
79 | | Status _prepare_fts_query_context(); |
80 | | Status _open_scanner(const TFileRangeDesc& range); |
81 | | void _init_scanner_profile(); |
82 | | Status _configure_scan_options(LanceScanner* scanner) const; |
83 | | Status _configure_normal_scan(LanceScanner* scanner, const TLanceFileDesc& lance_params) const; |
84 | | Status _configure_vector_search(LanceScanner* scanner, |
85 | | const TLanceFileDesc& lance_params) const; |
86 | | Status _configure_full_text_search(LanceScanner* scanner, |
87 | | const TLanceFileDesc& lance_params) const; |
88 | | // Collect the cumulative statistics owned by this dataset handle. The Lance-C API returns an |
89 | | // absolute snapshot, so this method replaces rather than increments the profile counters. |
90 | | void _collect_data_cache_statistics(); |
91 | | // Keep lance-c's anonymous statistics typedef out of this header. _open_scanner installs the |
92 | | // strongly typed C callback adapter before forwarding the borrowed value here. |
93 | | static void _collect_scan_statistics(void* callback_ctx, const void* opaque_statistics); |
94 | | void _close_scanner(); |
95 | | void _close_dataset(); |
96 | | Status _fill_block_from_lance_batch(LanceBatch* batch, Block* block, size_t* rows); |
97 | | Status _dataset_key(const TFileRangeDesc& range, DatasetKey* key) const; |
98 | | |
99 | | LanceDataset* _dataset = nullptr; |
100 | | std::shared_ptr<arrow::Schema> _dataset_schema; |
101 | | LanceScanner* _scanner = nullptr; |
102 | | ShardedKVCache* _runtime_filter_cache = nullptr; |
103 | | std::optional<DatasetKey> _opened_dataset_key; |
104 | | LanceRecordBatchConverter _record_batch_converter; |
105 | | size_t _scanner_batch_size = 0; |
106 | | RuntimeProfile::Counter* _planned_index_segment_count = nullptr; |
107 | | RuntimeProfile::Counter* _planned_indexed_fragment_count = nullptr; |
108 | | RuntimeProfile::Counter* _planned_flat_search_fragment_count = nullptr; |
109 | | RuntimeProfile::Counter* _dataset_open_time = nullptr; |
110 | | RuntimeProfile::Counter* _scanner_configure_time = nullptr; |
111 | | RuntimeProfile::Counter* _runtime_filter_sql_translation_time = nullptr; |
112 | | RuntimeProfile::Counter* _scanner_next_time = nullptr; |
113 | | RuntimeProfile::Counter* _arrow_to_doris_block_time = nullptr; |
114 | | RuntimeProfile::Counter* _row_id_take_read_time = nullptr; |
115 | | RuntimeProfile::Counter* _row_id_fetch_total_time = nullptr; |
116 | | RuntimeProfile::Counter* _execution_iops = nullptr; |
117 | | RuntimeProfile::Counter* _execution_requests = nullptr; |
118 | | RuntimeProfile::Counter* _execution_bytes_read = nullptr; |
119 | | // Pre-registered before installing Lance's callback so it only updates existing counters. |
120 | | std::vector<RuntimeProfile::Counter*> _scan_metric_counters; |
121 | | RuntimeProfile::Counter* _data_cache_bytes_read_from_cache = nullptr; |
122 | | RuntimeProfile::Counter* _data_cache_bytes_read_from_remote = nullptr; |
123 | | RuntimeProfile::Counter* _index_object_loads = nullptr; |
124 | | RuntimeProfile::Counter* _index_components_loaded = nullptr; |
125 | | RuntimeProfile::Counter* _index_comparisons = nullptr; |
126 | | LanceFtsQueryContext* _fts_query_context = nullptr; |
127 | | SearchKind _search_kind = SearchKind::NORMAL; |
128 | | bool _eof = false; |
129 | | }; |
130 | | |
131 | | } // namespace doris::format::lance |