20 #include <condition_variable>
24 #include <rapidjson/document.h>
25 #include <boost/filesystem.hpp>
36 namespace foreign_storage {
39 , foreign_table_(nullptr)
40 , user_mapping_(nullptr)
41 , disable_cache_(
false)
42 , is_first_file_scan_call_(
true)
43 , is_file_scan_in_progress_(
false)
44 , iterative_scan_last_fragment_id_(-1) {}
50 , foreign_table_(foreign_table)
52 , user_mapping_(nullptr)
53 , disable_cache_(
false)
54 , is_first_file_scan_call_(
true)
55 , is_file_scan_in_progress_(
false)
56 , iterative_scan_last_fragment_id_(-1) {}
62 const bool disable_cache)
64 , foreign_table_(foreign_table)
66 , user_mapping_(user_mapping)
67 , disable_cache_(disable_cache)
68 , is_first_file_scan_call_(
true)
69 , is_file_scan_in_progress_(
false)
70 , iterative_scan_last_fragment_id_(-1) {}
75 const int32_t fragment_id,
76 const int32_t max_fragment_id) {
78 "Attempting to populate fragment id " +
std::to_string(fragment_id) +
79 " for foreign table " + table->
tableName +
80 " which is greater than the maximum fragment id of " +
86 const int32_t table_id,
87 const int fragment_id) {
88 CHECK(!buffers.empty());
89 std::set<const ColumnDescriptor*> columns;
90 for (
const auto& entry : buffers) {
94 columns.emplace(column);
105 const std::set<const ColumnDescriptor*>& columns,
106 const int fragment_id,
108 std::map<int, Chunk_NS::Chunk>& column_id_to_chunk_map) {
109 for (
const auto column : columns) {
115 column_id_to_chunk_map[column->columnId]);
126 CHECK(!required_buffers.empty());
129 auto required_columns =
131 std::map<int, Chunk_NS::Chunk> column_id_to_chunk_map;
133 required_columns, fragment_id, required_buffers, column_id_to_chunk_map);
135 if (!optional_buffers.empty()) {
136 auto optional_columns =
139 optional_columns, fragment_id, optional_buffers, column_id_to_chunk_map);
141 populateChunks(column_id_to_chunk_map, fragment_id, delete_buffer);
149 std::map<int, Chunk_NS::Chunk>& column_id_to_chunk_map,
153 for (
auto& entry : column_id_to_chunk_map) {
159 if (column->columnType.is_varlen_indeed()) {
160 data_chunk_key.emplace_back(1);
165 auto cached_metadata_previous =
168 std::make_shared<ChunkMetadata>();
170 *cached_metadata = *cached_metadata_previous;
171 auto chunk_metadata =
172 entry.second.getBuffer()->getEncoder()->getMetadata(column->columnType);
173 cached_metadata->chunkStats.max = chunk_metadata->chunkStats.max;
174 cached_metadata->chunkStats.min = chunk_metadata->chunkStats.min;
175 cached_metadata->chunkStats.has_nulls = chunk_metadata->chunkStats.has_nulls;
176 cached_metadata->numBytes = entry.second.getBuffer()->size();
198 const size_t num_loaded,
199 const std::string& item_type,
200 const std::string& foreign_table_name) {
203 num_expected, num_loaded, item_type);
206 std::string(except.what()) +
" Foreign table: " + foreign_table_name);
218 const size_t start_index,
219 const size_t end_index,
222 const std::map<int, Chunk_NS::Chunk>& column_id_to_chunk_map,
226 load_file_region_result.
file_offset = file_regions[start_index].first_row_file_offset;
227 load_file_region_result.row_count = 0;
230 for (
size_t i = start_index; i <= end_index; i++) {
233 file_regions[i].first_row_file_offset,
234 file_regions[i].region_size);
235 if (file_regions[i].region_size != read_size) {
242 parse_file_request.
end_pos = file_regions[i].region_size;
244 parse_file_request.
file_offset = file_regions[i].first_row_file_offset;
247 result = parser.
parseBuffer(parse_file_request, i == end_index);
249 for (
const auto& rejected_row_index : result.
rejected_rows) {
250 load_file_region_result.rejected_row_indices.insert(
251 load_file_region_result.row_count + rejected_row_index);
253 load_file_region_result.row_count += result.
row_count;
255 load_file_region_result.column_id_to_data_blocks_map =
257 return load_file_region_result;
266 const bool size_known,
269 if (size_known && file_size < buffer_size) {
270 buffer_size = file_size + 1;
276 size_t buffer_size = 0;
277 for (
const auto& file_region : file_regions) {
278 buffer_size = std::max(buffer_size, file_region.region_size);
289 const bool size_known,
291 const size_t buffer_size) {
292 size_t thread_count = copy_params.
threads;
293 if (thread_count == 0) {
294 thread_count = std::thread::hardware_concurrency();
296 if (size_known && file_size > 0) {
297 size_t num_buffers_in_file = (file_size + buffer_size - 1) / buffer_size;
298 if (num_buffers_in_file < thread_count) {
299 thread_count = num_buffers_in_file;
302 CHECK_GT(thread_count, static_cast<size_t>(0));
308 size_t thread_count = copy_params.
threads;
309 if (thread_count == 0) {
311 std::min<size_t>(std::thread::hardware_concurrency(), file_regions.size());
313 CHECK_GT(thread_count, static_cast<size_t>(0));
318 const size_t chunk_element_count) {
319 if (delete_buffer->
size() < chunk_element_count) {
320 auto remaining_rows = chunk_element_count - delete_buffer->
size();
321 std::vector<int8_t> data(remaining_rows,
false);
322 delete_buffer->
append(data.data(), remaining_rows);
327 std::unique_lock<std::mutex> deferred_requests_lock(
340 std::map<int, Chunk_NS::Chunk>& column_id_to_chunk_map,
345 CHECK(!column_id_to_chunk_map.empty());
359 auto file_regions_it_one_ahead =
365 column_id_to_chunk_map, fragment_id, delete_buffer};
386 const auto& file_regions = file_regions_it->second;
389 if (file_regions.empty()) {
396 const int batch_size = (file_regions.size() + thread_count - 1) / thread_count;
398 std::vector<ParseBufferRequest> parse_file_requests{};
399 parse_file_requests.reserve(thread_count);
400 std::vector<std::future<ParseFileRegionResult>> futures{};
401 std::set<int> column_filter_set;
402 for (
const auto& pair : column_id_to_chunk_map) {
403 column_filter_set.insert(pair.first);
406 std::vector<std::unique_ptr<FileReader>> file_readers;
407 rapidjson::Value reader_metadata(rapidjson::kObjectType);
408 rapidjson::Document d;
410 file_reader_->serialize(reader_metadata, d.GetAllocator());
414 for (
size_t i = 0; i < file_regions.size(); i += batch_size) {
415 parse_file_requests.emplace_back(buffer_size,
422 delete_buffer !=
nullptr);
423 auto start_index = i;
425 std::min<size_t>(start_index + batch_size - 1, file_regions.size() - 1);
428 file_readers.emplace_back(std::make_unique<LocalMultiFileReader>(
429 file_path, copy_params, reader_metadata));
436 std::ref(file_regions),
439 std::ref(*(file_readers.back())),
440 std::ref(parse_file_requests.back()),
441 std::ref(column_id_to_chunk_map),
445 for (
auto& future : futures) {
449 std::vector<ParseFileRegionResult> load_file_region_results{};
450 for (
auto& future : futures) {
451 load_file_region_results.emplace_back(future.get());
454 std::set<size_t> chunk_rejected_row_indices;
455 size_t chunk_offset = 0;
456 for (
auto result : load_file_region_results) {
457 for (
auto& [column_id, chunk] : column_id_to_chunk_map) {
459 result.column_id_to_data_blocks_map[column_id],
result.row_count, 0);
461 for (
const auto& rejected_row_index :
result.rejected_row_indices) {
462 chunk_rejected_row_indices.insert(rejected_row_index + chunk_offset);
464 chunk_offset +=
result.row_count;
471 auto delete_buffer_data = delete_buffer->
getMemoryPtr();
472 for (
const auto rejected_row_index : chunk_rejected_row_indices) {
473 delete_buffer_data[rejected_row_index] =
true;
484 const size_t max_fragment_size,
485 const size_t rows_remaining) {
486 size_t start_position_in_fragment = start_row_index % max_fragment_size;
487 return std::min<size_t>(rows_remaining, max_fragment_size - start_position_in_fragment);
497 const size_t max_fragment_size,
498 const size_t buffer_row_count) {
499 CHECK(buffer_row_count > 0);
500 std::vector<size_t> partitions{};
501 size_t remaining_rows_in_last_fragment;
502 if (start_row_index % max_fragment_size == 0) {
503 remaining_rows_in_last_fragment = 0;
505 remaining_rows_in_last_fragment =
506 max_fragment_size - (start_row_index % max_fragment_size);
508 if (buffer_row_count <= remaining_rows_in_last_fragment) {
509 partitions.emplace_back(buffer_row_count);
511 if (remaining_rows_in_last_fragment > 0) {
512 partitions.emplace_back(remaining_rows_in_last_fragment);
514 size_t remaining_buffer_row_count =
515 buffer_row_count - remaining_rows_in_last_fragment;
516 while (remaining_buffer_row_count > 0) {
517 partitions.emplace_back(
518 std::min<size_t>(remaining_buffer_row_count, max_fragment_size));
519 remaining_buffer_row_count -= partitions.back();
531 std::unique_lock<std::mutex> pending_requests_lock(
534 pending_requests_lock, [&multi_threading_params] {
543 pending_requests_lock.unlock();
545 return std::move(request);
554 size_t first_row_index,
556 const std::string& file_path) {
557 fragment_id_to_file_regions_map[fragment_id].emplace_back(
573 const size_t row_count) {
584 std::shared_ptr<Catalog_Namespace::Catalog>& catalog,
585 const bool disable_cache) {
586 if (!disable_cache && catalog->getDataMgr()
587 .getPersistentStorageMgr()
588 ->getDiskCacheConfig()
589 .isEnabledForFSI()) {
590 return catalog->getDataMgr().getPersistentStorageMgr()->getDiskCache();
604 bool disable_cache) {
623 if (cached_chunks.find(chunk_key) == cached_chunks.end()) {
625 cached_chunks[chunk_key].setBuffer(
626 cache->getChunkBufferForPrecaching(chunk_key, is_first_block));
628 cached_chunks[chunk_key].setIndexBuffer(
629 cache->getChunkBufferForPrecaching(index_key, is_first_block));
631 if (is_first_block) {
632 cached_chunks[chunk_key].initEncoder();
635 cached_chunks[chunk_key].appendData(data_block, row_count, 0);
645 const size_t element_count_required) {
650 std::unique_lock<std::mutex> chunk_lock(file_scan_param.
getChunkMutex(column_id));
651 conditional_variable.wait(chunk_lock, [element_count_required, &chunk]() {
652 return chunk.getBuffer()->getEncoder()->getNumElems() == element_count_required;
655 chunk.appendData(data_block, row_count, 0);
670 const std::map<int, const ColumnDescriptor*>& column_by_id,
671 const std::map<int, DataBlockPtr>& data_blocks) {
672 std::map<int, DataBlockPtr> dict_encoded_data_blocks;
673 std::map<int, DataBlockPtr> none_dict_encoded_data_blocks;
674 for (
auto& [column_id, data_block] : data_blocks) {
676 if (column->columnType.is_dict_encoded_string()) {
677 dict_encoded_data_blocks[column_id] = data_block;
679 none_dict_encoded_data_blocks[column_id] = data_block;
682 return {dict_encoded_data_blocks, none_dict_encoded_data_blocks};
689 const size_t start_position_in_fragment) {
693 auto chunk_offset = start_position_in_fragment;
699 auto delete_buffer_data = delete_buffer->getMemoryPtr();
701 CHECK(rejected_row_index + chunk_offset < delete_buffer->size());
702 delete_buffer_data[rejected_row_index + chunk_offset] =
true;
712 std::map<int, const ColumnDescriptor*>& column_by_id,
713 std::map<int, FileRegions>& fragment_id_to_file_regions_map,
715 const size_t expected_current_element_count) {
726 std::vector<std::pair<const size_t, std::future<int8_t*>>>
727 encoded_data_block_ptrs_futures;
730 if (import_buffer ==
nullptr) {
734 if (import_buffer->getTypeInfo().is_dict_encoded_string()) {
735 auto string_payload_ptr = import_buffer->getStringBuffer();
738 auto column_id = import_buffer->getColumnDesc()->columnId;
739 encoded_data_block_ptrs_futures.emplace_back(std::make_pair(
741 import_buffer->addDictEncodedString(*string_payload_ptr);
742 return import_buffer->getStringDictBuffer();
747 auto process_subset_of_data_blocks =
748 [&](
const std::map<int, DataBlockPtr>& data_blocks) {
749 for (
auto& [column_id, data_block] : data_blocks) {
750 const auto column = column_by_id[column_id];
758 expected_current_element_count);
763 auto [dict_encoded_data_blocks, none_dict_encoded_data_blocks] =
766 process_subset_of_data_blocks(
767 none_dict_encoded_data_blocks);
770 for (
auto& encoded_ptr_future : encoded_data_block_ptrs_futures) {
771 encoded_ptr_future.second.wait();
773 for (
auto& encoded_ptr_future : encoded_data_block_ptrs_futures) {
774 CHECK_GT(dict_encoded_data_blocks.count(encoded_ptr_future.first), 0UL);
775 dict_encoded_data_blocks[encoded_ptr_future.first].numbersPtr =
776 encoded_ptr_future.second.get();
779 process_subset_of_data_blocks(
780 dict_encoded_data_blocks);
792 std::map<int, const ColumnDescriptor*>& column_by_id,
793 std::map<int, FileRegions>& fragment_id_to_file_regions_map) {
804 const auto column = column_by_id[column_id];
805 if (column->columnType.is_varlen_indeed()) {
806 chunk_key.emplace_back(1);
811 std::make_unique<ForeignStorageBuffer>();
842 std::unique_lock<std::mutex> completed_requests_queue_lock(
844 multi_threading_params.
request_pool.emplace(std::move(request));
845 completed_requests_queue_lock.unlock();
854 std::map<int, FileRegions>& fragment_id_to_file_regions_map,
856 std::map<int, const ColumnDescriptor*> column_by_id{};
859 if (!request_opt.has_value()) {
862 auto& request = request_opt.value();
864 if (column_by_id.empty()) {
865 for (
const auto column : request.getColumns()) {
866 column_by_id[column->columnId] = column;
870 request.first_row_index, request.getMaxFragRows(), request.buffer_row_count);
871 request.begin_pos = 0;
872 size_t row_index = request.first_row_index;
873 for (
const auto partition : partitions) {
874 request.process_row_count = partition;
875 for (
const auto& import_buffer : request.import_buffers) {
876 if (import_buffer !=
nullptr) {
877 import_buffer->clear();
881 int fragment_id = row_index / request.getMaxFragRows();
887 fragment_id_to_file_regions_map);
888 row_index +=
result.row_count;
889 request.begin_pos =
result.row_offsets.back() - request.file_offset;
894 std::lock_guard<std::mutex> pending_requests_lock(
910 std::unique_lock<std::mutex> request_pool_lock(
914 [&multi_threading_params] {
return !multi_threading_params.
request_pool.empty(); });
915 auto request = std::move(multi_threading_params.
request_pool.front());
917 request_pool_lock.unlock();
918 CHECK(request.buffer);
923 std::unique_lock<std::mutex> request_pool_lock(
935 std::unique_lock<std::mutex> deferred_requests_lock(
945 std::unique_lock<std::mutex> deferred_requests_lock(
948 std::unique_lock<std::mutex> pending_requests_lock(
967 std::unique_lock<std::mutex> pending_requests_lock(
979 std::map<int, FileRegions>& fragment_id_to_file_regions_map,
982 std::map<int, const ColumnDescriptor*> column_by_id{};
985 if (!request_opt.has_value()) {
990 if (column_by_id.empty()) {
991 for (
const auto column : request.
getColumns()) {
992 column_by_id[column->columnId] = column;
996 for (
size_t num_rows_left_to_process =
998 num_rows_left_to_process > 0;
999 num_rows_left_to_process =
1013 if (import_buffer !=
nullptr) {
1014 import_buffer->clear();
1018 size_t start_position_in_fragment = row_index % request.
getMaxFragRows();
1024 fragment_id_to_file_regions_map,
1026 start_position_in_fragment);
1032 request,
result, file_scan_param, start_position_in_fragment);
1038 std::lock_guard<std::mutex> pending_requests_lock(
1054 size_t& buffer_size,
1055 const size_t alloc_size) {
1057 if (buffer_size < alloc_size) {
1058 buffer = std::make_unique<char[]>(alloc_size);
1059 buffer_size = alloc_size;
1078 const size_t& buffer_size,
1079 const std::string& file_path,
1083 size_t& first_row_index_in_buffer,
1084 size_t& current_file_offset,
1088 iterative_residual_buffer,
1089 const bool is_first_file_scan_call,
1090 int& iterative_scan_last_fragment_id) {
1091 auto& alloc_size = iterative_residual_buffer.
alloc_size;
1092 auto& residual_buffer = iterative_residual_buffer.
residual_data;
1096 if (is_first_file_scan_call) {
1097 alloc_size = buffer_size;
1098 residual_buffer = std::make_unique<char[]>(alloc_size);
1099 residual_buffer_size = 0;
1100 residual_buffer_alloc_size = alloc_size;
1109 bool current_fragment_fully_read_during_iterative_scan =
1110 file_scan_param && file_scan_param->
fragment_id < iterative_scan_last_fragment_id;
1126 !current_fragment_fully_read_during_iterative_scan)) {
1128 std::lock_guard<std::mutex> pending_requests_lock(
1138 if (residual_buffer_size > 0) {
1139 memcpy(request.buffer.get(), residual_buffer.get(), residual_buffer_size);
1141 size_t size = residual_buffer_size;
1142 size += file_reader.
read(request.buffer.get() + residual_buffer_size,
1143 alloc_size - residual_buffer_size);
1151 }
else if (size == 1 && request.buffer[0] == copy_params.
line_delim) {
1155 current_file_offset++;
1159 unsigned int num_rows_in_buffer = 0;
1164 first_row_index_in_buffer,
1167 request.buffer_size = size;
1168 request.buffer_alloc_size = alloc_size;
1169 request.first_row_index = first_row_index_in_buffer;
1170 request.file_offset = current_file_offset;
1171 request.buffer_row_count = num_rows_in_buffer;
1172 request.processed_row_count = 0;
1173 request.begin_pos = 0;
1175 residual_buffer_size = size - request.end_pos;
1176 if (residual_buffer_size > 0) {
1178 memcpy(residual_buffer.get(),
1179 request.buffer.get() + request.end_pos,
1180 residual_buffer_size);
1183 current_file_offset += request.end_pos;
1184 first_row_index_in_buffer += num_rows_in_buffer;
1186 if (num_rows_in_buffer > 0) {
1192 if (file_scan_param) {
1193 const int32_t last_fragment_index =
1195 if (last_fragment_index > file_scan_param->
fragment_id) {
1196 iterative_scan_last_fragment_id = last_fragment_index;
1202 std::unique_lock<std::mutex> pending_requests_queue_lock(
1205 pending_requests_queue_lock, [&multi_threading_params] {
1210 pending_requests_queue_lock.unlock();
1216 const size_t& buffer_size,
1217 const std::string& file_path,
1221 size_t& first_row_index_in_buffer,
1222 size_t& current_file_offset,
1226 iterative_residual_buffer,
1227 const bool is_first_file_scan_call,
1228 int& iterative_scan_last_fragment_id) {
1235 multi_threading_params,
1236 first_row_index_in_buffer,
1237 current_file_offset,
1240 iterative_residual_buffer,
1241 is_first_file_scan_call,
1242 iterative_scan_last_fragment_id);
1245 std::unique_lock<std::mutex> pending_requests_lock(
1256 const size_t& buffer_size,
1257 const std::string& file_path,
1261 size_t& first_row_index_in_buffer,
1262 size_t& current_file_offset,
1266 iterative_residual_buffer,
1267 const bool is_first_file_scan_call) {
1274 multi_threading_params,
1275 first_row_index_in_buffer,
1276 current_file_offset,
1279 iterative_residual_buffer,
1280 is_first_file_scan_call,
1293 const size_t start_row,
1294 const size_t total_num_rows,
1295 std::map<
ChunkKey, std::shared_ptr<ChunkMetadata>>& chunk_metadata_map) {
1298 chunk_key.emplace_back(1);
1302 int start_fragment = start_row / foreign_table->
maxFragRows;
1303 int end_fragment{0};
1304 if (total_num_rows > 0) {
1305 end_fragment = (total_num_rows - 1) / foreign_table->
maxFragRows;
1307 for (
int fragment_id = start_fragment; fragment_id <= end_fragment; fragment_id++) {
1308 size_t num_elements = (
static_cast<size_t>(foreign_table->
maxFragRows *
1309 (fragment_id + 1)) > total_num_rows)
1314 chunk_metadata_map[chunk_key] =
1320 const std::map<
ChunkKey, std::shared_ptr<ChunkMetadata>>& chunk_metadata_map,
1321 const std::map<int, FileRegions>& fragment_id_to_file_regions_map,
1323 std::unique_ptr<FileReader>& file_reader,
1324 const std::string& file_path,
1327 const std::optional<size_t>& max_file_count,
1331 std::function<std::string()> get_s3_key,
1333 size_t& append_start_offset) {
1335 CHECK(chunk_metadata_map.empty());
1336 CHECK(fragment_id_to_file_regions_map.empty());
1340 file_reader = std::make_unique<LocalMultiFileReader>(
1341 file_path, copy_params, file_path_options, max_file_count);
1347 append_start_offset = 0;
1374 std::set<std::string> rolled_off_files;
1406 std::map<int32_t, const ColumnDescriptor*> column_by_id{};
1407 for (
auto column : columns) {
1408 column_by_id[column->columnId] = column;
1418 std::set<int> columns_to_scan;
1419 for (
auto column : columns) {
1421 columns_to_scan.insert(column->columnId);
1437 std::vector<std::future<void>> futures{};
1438 for (
size_t i = 0; i < thread_count; i++) {
1439 multi_threading_params.
request_pool.emplace(buffer_size,
1450 std::ref(multi_threading_params),
1461 multi_threading_params,
1469 for (
auto& future : futures) {
1477 CHECK(column_entry != column_by_id.end());
1478 const auto& column_type = column_entry->second->columnType;
1479 auto chunk_metadata = buffer->getEncoder()->getMetadata(column_type);
1480 chunk_metadata->numElements = buffer->getEncoder()->getNumElems();
1481 const auto& cached_chunks = multi_threading_params.
cached_chunks;
1482 if (!column_type.is_varlen_indeed()) {
1483 chunk_metadata->numBytes = column_type.get_size() * chunk_metadata->numElements;
1484 }
else if (
auto chunk_entry = cached_chunks.find(chunk_key);
1485 chunk_entry != cached_chunks.end()) {
1486 auto buffer = chunk_entry->second.getBuffer();
1488 chunk_metadata->numBytes = buffer->size();
1490 CHECK_EQ(chunk_metadata->numBytes, static_cast<size_t>(0));
1495 for (
auto column : columns) {
1502 if (!rolled_off_files.empty()) {
1507 chunk_metadata_vector.emplace_back(chunk_key, chunk_metadata);
1524 <<
" iterative file scan can not be used with APPEND mode.";
1555 std::map<int32_t, const ColumnDescriptor*> column_by_id{};
1556 for (
auto column : columns) {
1557 column_by_id[column->columnId] = column;
1567 std::set<int> columns_to_scan;
1568 for (
auto column : columns) {
1569 columns_to_scan.insert(column->columnId);
1586 std::vector<std::future<void>> futures{};
1603 std::ref(file_scan_param)));
1620 for (
auto& future : futures) {
1636 const std::set<std::string>& rolled_off_files,
1637 const std::map<int32_t, const ColumnDescriptor*>& column_by_id) {
1638 std::set<int32_t> deleted_fragment_ids;
1639 std::optional<int32_t> partially_deleted_fragment_id;
1640 std::optional<size_t> partially_deleted_fragment_row_count;
1642 bool file_region_deleted{
false};
1643 for (
auto it = file_regions.begin(); it != file_regions.end();) {
1645 it = file_regions.erase(it);
1646 file_region_deleted =
true;
1651 if (file_regions.empty()) {
1652 deleted_fragment_ids.emplace(fragment_id);
1653 }
else if (file_region_deleted) {
1654 partially_deleted_fragment_id = fragment_id;
1655 partially_deleted_fragment_row_count = 0;
1656 for (
const auto& file_region : file_regions) {
1657 partially_deleted_fragment_row_count.value() += file_region.row_count;
1665 chunk_metadata->numElements = 0;
1666 chunk_metadata->numBytes = 0;
1667 }
else if (chunk_key[CHUNK_KEY_FRAGMENT_IDX] == partially_deleted_fragment_id) {
1668 CHECK(partially_deleted_fragment_row_count.has_value());
1669 auto old_chunk_stats = chunk_metadata->chunkStats;
1672 cd->columnType, partially_deleted_fragment_row_count.value());
1675 chunk_metadata->chunkStats = old_chunk_stats;
1681 rapidjson::Document d;
1687 "fragment_id_to_file_regions_map",
1691 rapidjson::Value reader_metadata(rapidjson::kObjectType);
1692 file_reader_->serialize(reader_metadata, d.GetAllocator());
1693 d.AddMember(
"reader_metadata", reader_metadata, d.GetAllocator());
1703 const std::string& file_path,
1706 CHECK(d.IsObject());
1713 CHECK(d.HasMember(
"reader_metadata"));
1719 full_file_path, copy_params, d[
"reader_metadata"]);
1731 for (
auto& pair : chunk_metadata) {
1740 pair.second->numElements);
1742 pair.second->chunkStats);
1766 for (
auto const& column : columns) {
1769 .try_emplace(column->columnId,
1770 std::make_unique<import_export::RenderGroupAnalyzer>())
std::map< int, DataBlockPtr > column_id_to_data_blocks_map
bool contains(const T &container, const U &element)
void append_data_block_to_chunk(const foreign_storage::IterativeFileScanParameters &file_scan_param, DataBlockPtr data_block, size_t row_count, const int column_id, const ColumnDescriptor *column, const size_t element_count_required)
std::vector< std::unique_ptr< import_export::TypedImportBuffer > > import_buffers
virtual bool isScanFinished() const =0
RenderGroupAnalyzerMap render_group_analyzer_map_
std::vector< int > ChunkKey
static shared::FilePathOptions getFilePathOptions(const ForeignTable *foreign_table)
virtual void validateFiles(const FileReader *file_reader, const ForeignTable *foreign_table) const =0
virtual size_t read(void *buffer, size_t max_size)=0
std::string getTableName() const
foreign_storage::ForeignStorageCache * get_cache_if_enabled(std::shared_ptr< Catalog_Namespace::Catalog > &catalog, const bool disable_cache)
class for a per-database catalog. also includes metadata for the current database and the current use...
void update_stats(Encoder *encoder, const SQLTypeInfo &column_type, DataBlockPtr data_block, const size_t row_count)
std::vector< std::string > * stringsPtr
std::set< size_t > rejected_rows
std::vector< ArrayDatum > * arraysPtr
void resize_delete_buffer(AbstractBuffer *delete_buffer, const size_t chunk_element_count)
std::pair< std::map< int, DataBlockPtr >, std::map< int, DataBlockPtr > > partition_data_blocks(const std::map< int, const ColumnDescriptor * > &column_by_id, const std::map< int, DataBlockPtr > &data_blocks)
void populate_chunks_using_data_blocks(MetadataScanMultiThreadingParams &multi_threading_params, int fragment_id, const ParseBufferRequest &request, ParseBufferResult &result, std::map< int, const ColumnDescriptor * > &column_by_id, std::map< int, FileRegions > &fragment_id_to_file_regions_map, const foreign_storage::IterativeFileScanParameters &file_scan_param, const size_t expected_current_element_count)
static const std::string LOCAL_FILE_STORAGE_TYPE
std::condition_variable & getChunkConditionalVariable(const int col_id) const
std::map< int, DataBlockPtr > column_id_to_data_blocks_map
#define CHUNK_KEY_FRAGMENT_IDX
void updateRolledOffChunks(const std::set< std::string > &rolled_off_files, const std::map< int32_t, const ColumnDescriptor * > &column_by_id)
const UserMapping * user_mapping_
virtual ParseBufferResult parseBuffer(ParseBufferRequest &request, bool convert_data_blocks, bool columns_are_pre_filtered=false, bool skip_dict_encoding=false) const =0
bool skip_metadata_scan(const ColumnDescriptor *column)
virtual int8_t * getMemoryPtr()=0
std::string getSerializedDataWrapper() const override
std::set< const ColumnDescriptor * > get_columns(const ChunkToBufferMap &buffers, const Catalog_Namespace::Catalog &catalog, const int32_t table_id, const int fragment_id)
virtual size_t readRegion(void *buffer, size_t offset, size_t size)=0
std::map< ChunkKey, AbstractBuffer * > ChunkToBufferMap
std::map< ChunkKey, std::unique_ptr< ForeignStorageBuffer > > chunk_encoder_buffers_
void reset_multithreading_params(foreign_storage::MetadataScanMultiThreadingParams &multi_threading_params)
std::vector< size_t > partition_by_fragment(const size_t start_row_index, const size_t max_fragment_size, const size_t buffer_row_count)
void add_placeholder_metadata(const ColumnDescriptor *column, const ForeignTable *foreign_table, const int db_id, const size_t start_row, const size_t total_num_rows, std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &chunk_metadata_map)
size_t num_rows_to_process(const size_t start_row_index, const size_t max_fragment_size, const size_t rows_remaining)
virtual const TextFileBufferParser & getFileBufferParser() const =0
std::map< int, Chunk_NS::Chunk > & column_id_to_chunk_map
size_t append_start_offset_
void dispatch_scan_requests(const foreign_storage::ForeignTable *table, const size_t &buffer_size, const std::string &file_path, FileReader &file_reader, const import_export::CopyParams ©_params, MetadataScanMultiThreadingParams &multi_threading_params, size_t &first_row_index_in_buffer, size_t ¤t_file_offset, const TextFileBufferParser &parser, const foreign_storage::IterativeFileScanParameters *file_scan_param, foreign_storage::AbstractTextFileDataWrapper::ResidualBuffer &iterative_residual_buffer, const bool is_first_file_scan_call, int &iterative_scan_last_fragment_id)
void add_request_to_pool(MetadataScanMultiThreadingParams &multi_threading_params, ParseBufferRequest &request)
size_t residual_buffer_alloc_size
bool no_deferred_requests(MetadataScanMultiThreadingParams &multi_threading_params)
const bool disable_cache_
std::set< size_t > rejected_row_indices
bool is_first_file_scan_call_
void iterativeFileScan(ChunkMetadataVector &chunk_metadata_vector, IterativeFileScanParameters &file_scan_param)
void createRenderGroupAnalyzers() override
Create RenderGroupAnalyzers for poly columns.
void populateChunkMetadata(ChunkMetadataVector &chunk_metadata_vector) override
size_t get_buffer_size(const import_export::CopyParams ©_params, const bool size_known, const size_t file_size)
std::vector< FileRegion > FileRegions
MetadataScanMultiThreadingParams multi_threading_params_
void get_value_from_object(const rapidjson::Value &object, T &value, const std::string &name)
bool key_does_not_shard_to_leaf(const ChunkKey &key)
std::shared_ptr< ChunkMetadata > get_placeholder_metadata(const SQLTypeInfo &type, size_t num_elements)
future< Result > async(Fn &&fn, Args &&...args)
static SysCatalog & instance()
static const std::string STORAGE_TYPE_KEY
size_t get_thread_count(const import_export::CopyParams ©_params, const bool size_known, const size_t file_size, const size_t buffer_size)
void throw_fragment_id_out_of_bounds_error(const TableDescriptor *table, const int32_t fragment_id, const int32_t max_fragment_id)
void update_delete_buffer(const ParseBufferRequest &request, const ParseBufferResult &result, const foreign_storage::IterativeFileScanParameters &file_scan_param, const size_t start_position_in_fragment)
std::map< ChunkKey, std::shared_ptr< ChunkMetadata > > chunk_metadata_map_
void populate_chunks(MetadataScanMultiThreadingParams &multi_threading_params, std::map< int, FileRegions > &fragment_id_to_file_regions_map, const TextFileBufferParser &parser, foreign_storage::IterativeFileScanParameters &file_scan_param)
void updateMetadata(std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id)
rapidjson::Document read_from_file(const std::string &file_path)
ParseFileRegionResult parse_file_regions(const FileRegions &file_regions, const size_t start_index, const size_t end_index, FileReader &file_reader, ParseBufferRequest &parse_file_request, const std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, const TextFileBufferParser &parser)
virtual std::optional< size_t > getMaxFileCount() const
std::unique_ptr< FileReader > file_reader_
void scan_metadata(MetadataScanMultiThreadingParams &multi_threading_params, std::map< int, FileRegions > &fragment_id_to_file_regions_map, const TextFileBufferParser &parser)
std::unique_lock< T > unique_lock
const ColumnDescriptor * getMetadataForColumn(int tableId, const std::string &colName) const
#define CHUNK_KEY_TABLE_IDX
void initialize_non_append_mode_scan(const std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &chunk_metadata_map, const std::map< int, FileRegions > &fragment_id_to_file_regions_map, const foreign_storage::OptionsMap &server_options, std::unique_ptr< FileReader > &file_reader, const std::string &file_path, const import_export::CopyParams ©_params, const shared::FilePathOptions &file_path_options, const std::optional< size_t > &max_file_count, const foreign_storage::ForeignTable *foreign_table, const foreign_storage::UserMapping *user_mapping, const foreign_storage::TextFileBufferParser &parser, std::function< std::string()> get_s3_key, size_t &num_rows, size_t &append_start_offset)
void init_chunk_for_column(const ChunkKey &chunk_key, const std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &chunk_metadata_map, const std::map< ChunkKey, AbstractBuffer * > &buffers, Chunk_NS::Chunk &chunk)
bool is_dict_encoded_type() const
An AbstractBuffer is a unit of data management for a data manager.
bool operator<(const ParseFileRegionResult &other) const
void dispatch_all_deferred_requests(MetadataScanMultiThreadingParams &multi_threading_params)
specifies the content in-memory of a row in the column metadata table
std::list< const ColumnDescriptor * > getColumns() const
bool g_enable_smem_group_by true
void add_value_to_object(rapidjson::Value &object, const T &value, const std::string &name, rapidjson::Document::AllocatorType &allocator)
void dispatch_scan_requests_with_exception_handling(const foreign_storage::ForeignTable *table, const size_t &buffer_size, const std::string &file_path, FileReader &file_reader, const import_export::CopyParams ©_params, MetadataScanMultiThreadingParams &multi_threading_params, size_t &first_row_index_in_buffer, size_t ¤t_file_offset, const TextFileBufferParser &parser, const foreign_storage::IterativeFileScanParameters *file_scan_param, foreign_storage::AbstractTextFileDataWrapper::ResidualBuffer &iterative_residual_buffer, const bool is_first_file_scan_call, int &iterative_scan_last_fragment_id)
bool isAppendMode() const
Checks if the table is in append mode.
std::unique_ptr< char[]> residual_data
std::mutex & getChunkMutex(const int col_id) const
size_t processed_row_count
bool request_pool_non_empty(MetadataScanMultiThreadingParams &multi_threading_params)
void defer_scan_request(MetadataScanMultiThreadingParams &multi_threading_params, ParseBufferRequest &request)
std::vector< size_t > row_offsets
std::shared_ptr< Catalog > getCatalog(const std::string &dbName)
static bool allowFileRollOff(const ForeignTable *foreign_table)
bool is_file_scan_in_progress_
V & get_from_map(std::map< K, V, comp > &map, const K &key)
void process_data_blocks(MetadataScanMultiThreadingParams &multi_threading_params, int fragment_id, const ParseBufferRequest &request, ParseBufferResult &result, std::map< int, const ColumnDescriptor * > &column_by_id, std::map< int, FileRegions > &fragment_id_to_file_regions_map)
const ForeignTable * foreign_table_
ParseBufferRequest get_request_from_pool(MetadataScanMultiThreadingParams &multi_threading_params)
size_t residual_buffer_size
void populateChunks(std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id, AbstractBuffer *delete_buffer)
ResidualBuffer residual_buffer_
bool isRestored() const override
int iterative_scan_last_fragment_id_
AbstractBuffer * delete_buffer
virtual void append(int8_t *src, const size_t num_bytes, const MemoryLevel src_buffer_type=CPU_LEVEL, const int device_id=-1)=0
void populateChunkBuffers(const ChunkToBufferMap &required_buffers, const ChunkToBufferMap &optional_buffers, AbstractBuffer *delete_buffer) override
void restoreDataWrapperInternals(const std::string &file_path, const ChunkMetadataVector &chunk_metadata) override
std::map< int, FileRegions > fragment_id_to_file_regions_map_
const ForeignServer * foreign_server
virtual size_t findRowEndPosition(size_t &alloc_size, std::unique_ptr< char[]> &buffer, size_t &buffer_size, const import_export::CopyParams ©_params, const size_t buffer_first_row_index, unsigned int &num_rows_in_buffer, FileReader *file_reader) const =0
bool g_enable_watchdog false
void add_file_region(std::map< int, FileRegions > &fragment_id_to_file_regions_map, int fragment_id, size_t first_row_index, const ParseBufferResult &result, const std::string &file_path)
#define DEBUG_TIMER(name)
std::map< std::string, std::string, std::less<>> OptionsMap
virtual import_export::CopyParams validateAndGetCopyParams(const ForeignTable *foreign_table) const =0
std::string write_to_string(const rapidjson::Document &document)
void populateChunkMapForColumns(const std::set< const ColumnDescriptor * > &columns, const int fragment_id, const ChunkToBufferMap &buffers, std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map)
int32_t getTableId() const
bool is_varlen_indeed() const
std::optional< ParseBufferRequest > get_next_scan_request(MetadataScanMultiThreadingParams &multi_threading_params)
#define CHUNK_KEY_COLUMN_IDX
bool is_file_scan_finished(const FileReader *file_reader, MetadataScanMultiThreadingParams &multi_threading_params)
size_t getMaxFragRows() const
void throw_unexpected_number_of_items(const size_t &num_expected, const size_t &num_loaded, const std::string &item_type)
virtual std::set< std::string > checkForRolledOffFiles(const shared::FilePathOptions &file_path_options)
std::mutex delete_buffer_mutex
std::string getFilePath() const
static std::string getFullFilePath(const ForeignTable *foreign_table)
Returns the path to the source file/dir of the table. Depending on options this may result from a con...
virtual void updateStats(const int64_t val, const bool is_null)=0
size_t file_size(const int fd)
std::unique_ptr< char[]> buffer
void resize_buffer_if_needed(std::unique_ptr< char[]> &buffer, size_t &buffer_size, const size_t alloc_size)
void dispatch_scan_request(MetadataScanMultiThreadingParams &multi_threading_params, ParseBufferRequest &request)
void cache_blocks(std::map< ChunkKey, Chunk_NS::Chunk > &cached_chunks, DataBlockPtr data_block, size_t row_count, ChunkKey &chunk_key, const ColumnDescriptor *column, bool is_first_block, bool disable_cache)
virtual std::string getCurrentFilePath() const =0
AbstractTextFileDataWrapper()