Changed-lines coverage: PR changed C/C++ lines covered by tests: 90.06% (652/724) Uncovered changed code (with context): ================================================================================ src/Interpreters/ClusterFunctionReadTask.cpp ================================================================================ --- uncovered block 117-117 --- 115 | /// Only the fail-close overwrite guard is lost, which a worker below the required protocol could 116 | /// not run anyway. >> 117 | bucket_info_to_send = nullptr; 118 | } 119 | ================================================================================ src/Processors/Formats/IInputFormat.cpp ================================================================================ --- uncovered block 16-16 --- 14 | UInt64 FileBucketInfo::getMinProtocolVersion() const 15 | { >> 16 | return DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_FILE_BUCKETS_INFO; 17 | } 18 | ================================================================================ src/Processors/Formats/Impl/Parquet/Reader.cpp ================================================================================ --- uncovered block 633-637 --- 631 | for (UInt64 rg : *row_groups_to_read) 632 | if (rg >= file_metadata.row_groups.size()) >> 633 | throw Exception( >> 634 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 635 | "Row group {} from the bucket assignment is out of range: the file has only {} row groups. " >> 636 | "The file was likely modified concurrently while a parallel single-file read was in progress", >> 637 | rg, file_metadata.row_groups.size()); 638 | } 639 | ================================================================================ src/Processors/Formats/Impl/ParquetV3BlockInputFormat.cpp ================================================================================ --- uncovered block 59-63 --- 57 | { 58 | if (bucket.file_num_row_groups != 0 && file_metadata.row_groups.size() != bucket.file_num_row_groups) >> 59 | throw Exception( >> 60 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 61 | "The Parquet file has {} row groups, but the parallel single-file bucket assignment was computed for a file " >> 62 | "with {} row groups. The file was likely modified concurrently while a parallel single-file read was in progress", >> 63 | file_metadata.row_groups.size(), bucket.file_num_row_groups); 64 | 65 | /// A matching row-group count can still hide a rewrite: on the local `StorageFile` path an --- uncovered block 72-75 --- 70 | /// case (see `ParquetFileBucketInfo::footer_digest`). 71 | if (bucket.footer_digest != 0 && computeParquetFooterDigest(file_metadata) != bucket.footer_digest) >> 72 | throw Exception( >> 73 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 74 | "The Parquet file's footer differs from the one the parallel single-file bucket assignment was computed from. " >> 75 | "The file was likely modified concurrently while a parallel single-file read was in progress"); 76 | } 77 | --- uncovered block 197-201 --- 195 | { 196 | if (rg >= file_metadata.row_groups.size()) >> 197 | throw Exception( >> 198 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 199 | "Row group {} from the bucket assignment is out of range: the file has only {} row groups. " >> 200 | "The file was likely modified concurrently while a parallel single-file read was in progress", >> 201 | rg, file_metadata.row_groups.size()); 202 | num_rows += size_t(file_metadata.row_groups[rg].num_rows); 203 | } --- uncovered block 366-366 --- 364 | const size_t result_file_num_row_groups = caller_file_num_row_groups != 0 ? caller_file_num_row_groups : file_num_row_groups; 365 | if (matching_row_groups.empty()) >> 366 | return nullptr; 367 | if (row_group_ids.empty()) 368 | { --- uncovered block 382-382 --- 380 | filtered.push_back(rg); 381 | if (filtered.empty()) >> 382 | return nullptr; 383 | /// Filtering a real split bucket: the row groups left out are a mix of the other buckets' row 384 | /// groups and the ones the cache dropped, so this reader stays accountable for its own bucket --- uncovered block 498-498 --- 496 | { 497 | if (target_count == 0 || num_row_groups == 0) >> 498 | return {}; 499 | 500 | static constexpr size_t min_row_groups_per_chunk = 16; --- uncovered block 535-535 --- 533 | { 534 | if (idx >= schema.size()) >> 535 | return false; 536 | const parquet::format::SchemaElement & elem = schema[idx]; 537 | ++idx; --- uncovered block 586-586 --- 584 | } 585 | /// Two-level list (e.g. hudi): the repeated element itself is the wrapper level. >> 586 | return collectLogicalPaths(schema, idx, raw, logical, false, out); 587 | } 588 | /// A MAP-annotated group without the expected key/value structure: fall through and treat --- uncovered block 595-595 --- 593 | for (size_t i = 0; i < num_children; ++i) 594 | if (!collectLogicalPaths(schema, idx, raw, logical, true, out)) >> 595 | return false; 596 | return true; 597 | } --- uncovered block 639-639 --- 637 | ok = collectLogicalPaths(md.schema, idx, "", "", true, logical_paths); 638 | if (!ok) >> 639 | logical_paths.clear(); 640 | } 641 | --- uncovered block 648-651 --- 646 | { 647 | if (!col.__isset.meta_data) >> 648 | continue; 649 | const auto & path = col.meta_data.path_in_schema; 650 | if (path.empty()) >> 651 | continue; 652 | if (!requested_columns.empty()) 653 | { --- uncovered block 666-666 --- 664 | matched = anyDottedPrefixRequested(it->second, requested_columns); 665 | else >> 666 | matched = requested_columns.contains(leaf_path); 667 | } 668 | if (!matched) --- uncovered block 696-696 --- 694 | { 695 | if (target_count == 0 || num_row_groups == 0) >> 696 | return {}; 697 | const bool apply_row_group_floor = min_bytes_to_split > 0 || min_bytes_per_bucket > 0; 698 | if (min_bytes_to_split > 0 && projected_bytes < min_bytes_to_split) --- uncovered block 721-721 --- 719 | { 720 | if (buckets.empty()) >> 721 | return; 722 | const UInt64 digest = computeParquetFooterDigest(file_metadata); 723 | for (auto & bucket : buckets) --- uncovered block 782-782 --- 780 | hash.update(column.__isset.file_path); 781 | if (column.__isset.file_path) >> 782 | hash.update(column.file_path); 783 | update_optional(column.__isset.offset_index_offset, column.offset_index_offset); 784 | update_optional(column.__isset.offset_index_length, column.offset_index_length); --- uncovered block 789-789 --- 787 | hash.update(column.__isset.meta_data); 788 | if (!column.__isset.meta_data) >> 789 | continue; 790 | const auto & meta = column.meta_data; 791 | hash.update(meta.num_values); --- uncovered block 812-815 --- 810 | std::vector ParquetBucketSplitter::splitToBucketsByCount(size_t target_count, ReadBuffer & buf, const FormatSettings & format_settings_) 811 | { >> 812 | auto file_metadata = parseFileMetadataNative(buf, format_settings_); >> 813 | auto buckets = computeBucketsByCount(target_count, file_metadata.row_groups.size()); >> 814 | setFooterDigest(buckets, file_metadata); >> 815 | return buckets; 816 | } 817 | ================================================================================ src/Storages/ObjectStorage/IObjectIterator.cpp ================================================================================ --- uncovered block 179-182 --- 177 | /// The bucket produced by the splitter already carries the file's total row-group 178 | /// count, so pass 0 ("unknown") here to keep it rather than overwrite it. >> 179 | auto filtered = file_bucket->filterByMatchingRowGroups(matching_row_groups, /*file_num_row_groups=*/ 0); >> 180 | if (!filtered) >> 181 | continue; >> 182 | copy_object_info.file_bucket_info = std::move(filtered); 183 | } 184 | else ================================================================================ src/Storages/ObjectStorage/StorageObjectStorageSource.cpp ================================================================================ --- uncovered block 921-921 --- 919 | auto filtered = file_bucket_info->filterByMatchingRowGroups(matching_row_groups, total_row_groups); 920 | if (!filtered) >> 921 | continue; 922 | object_info->file_bucket_info = std::move(filtered); 923 | } ================================================================================ src/Storages/ObjectStorage/tests/gtest_rendezvous_hashing.cpp ================================================================================ --- uncovered block 100-103 --- 98 | : min_protocol_version(min_protocol_version_) {} 99 | void serialize(WriteBuffer &, size_t) override {} >> 100 | void deserialize(ReadBuffer &, size_t) override {} >> 101 | String getIdentifier() const override { return "stub"; } 102 | String getFormatName() const override { return "Parquet"; } >> 103 | std::shared_ptr filterByMatchingRowGroups(const std::vector &, size_t) const override { return nullptr; } 104 | UInt64 getMinProtocolVersion() const override { return min_protocol_version; } 105 | UInt64 min_protocol_version; ================================================================================ src/Storages/StorageFile.cpp ================================================================================ --- uncovered block 1776-1780 --- 1774 | if (file_bucket_info && expected_file_cache_version.has_value() 1775 | && *expected_file_cache_version != *current_file_cache_version) >> 1776 | throw Exception( >> 1777 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 1778 | "File {} was modified concurrently while a parallel single-file read was in progress " >> 1779 | "(version changed from {} to {})", >> 1780 | current_path, *expected_file_cache_version, *current_file_cache_version); 1781 | 1782 | /// The version token above proves a rewrite only after the file has settled --- uncovered block 1826-1838 --- 1824 | /// changed under us could silently return wrong data, so fail close, mirroring 1825 | /// the pre-open check above. >> 1826 | if (file_bucket_info && expected_file_cache_version.has_value()) >> 1827 | throw Exception( >> 1828 | ErrorCodes::FILE_CHANGED_WHILE_READING, >> 1829 | "File {} was modified concurrently while a parallel single-file read was in progress " >> 1830 | "(version {} did not hold after the file was opened)", >> 1831 | current_path, *current_file_cache_version); 1832 | 1833 | /// For a plain read, fail close on the caches instead: drop the token so neither 1834 | /// the format metadata cache (`object_with_metadata` below) nor the Query 1835 | /// Condition Cache keys this read's data under a version it may not describe. 1836 | /// The footer is then parsed directly from the opened bytes. >> 1837 | current_file_cache_version.reset(); >> 1838 | current_file_version_settled = false; 1839 | 1840 | /// The size and mtime also come from the pre-open `stat`, and the `_size` / --- uncovered block 1845-1846 --- 1843 | /// to the opened bytes anymore, so fail close here too: with the optionals 1844 | /// empty the virtual columns are filled with NULL. >> 1845 | current_file_size.reset(); >> 1846 | current_file_last_modified.reset(); 1847 | } 1848 | } --- uncovered block 1968-1968 --- 1966 | if (!buckets_to_read) 1967 | { >> 1968 | read_buf.reset(); 1969 | continue; 1970 | } --- uncovered block 2481-2482 --- 2479 | if (info.row_level_filter) 2480 | { >> 2481 | for (const auto & input : info.row_level_filter->actions.getRequiredColumns()) >> 2482 | requested_columns.insert(input.name); 2483 | } 2484 | if (info.prewhere_info) --- uncovered block 2518-2524 --- 2516 | } 2517 | else >> 2518 | #endif 2519 | { >> 2520 | auto buf = createReadBuffer( >> 2521 | single_file_path, file_stat, false, -1, storage->compression_method, ctx); >> 2522 | const auto & format_settings = storage->format_settings.value_or(getFormatSettings(ctx)); >> 2523 | auto splitter = FormatFactory::instance().getSplitter(storage->format_name); >> 2524 | buckets = splitter->splitToBucketsByCount(max_num_streams, *buf, format_settings); 2525 | } 2526 | --- uncovered block 2536-2536 --- 2534 | /// read handles fine). 2535 | if (!buckets.empty() && !fileCacheVersionTokenStillHolds(single_file_path, decision_file_version)) >> 2536 | buckets.clear(); 2537 | 2538 | if (buckets.size() >= 2) WARNING: Failed to get start time for [Print Uncovered Code] - start time and duration won't be set --- Coverage counts --- Lines : baseline 943,198/1,089,326 -> current 943,485/1,090,035 (delta +287 / +709) Functions : baseline 808,554/880,119 -> current 808,605/880,200 (delta +51 / +81) Branches : baseline 310,002/393,144 -> current 310,218/393,554 (delta +216 / +410)