| 785 | } |
| 786 | |
| 787 | void DiskObjectStorage::prepareRead( |
| 788 | const String & path, |
| 789 | const ReadSettings & settings, |
| 790 | std::optional<size_t> read_hint, |
| 791 | ReadPipeline & pipeline) const |
| 792 | { |
| 793 | const auto storage_objects = metadata_storage->getStorageObjects(path); |
| 794 | |
| 795 | auto read_settings = updateIOSchedulingSettings(settings, getReadResourceName(), getWriteResourceName()); |
| 796 | auto global_context = Context::getGlobalContextInstance(); |
| 797 | auto storage = object_storages->takePointingTo(cluster->getLocalLocation()); |
| 798 | |
| 799 | /// Empty objects (zero-blob file) — set an empty source so `ReadPipeline::build` |
| 800 | /// returns `ReadBufferFromEmptyFile`. No stages are needed below the source. |
| 801 | if (storage_objects.empty()) |
| 802 | { |
| 803 | pipeline.setSource(std::move(storage), StoredObjects{}, settings); |
| 804 | return; |
| 805 | } |
| 806 | |
| 807 | /// Distributed cache — computed early because prefer_bigger_buffer_size needs to be known. |
| 808 | #if ENABLE_DISTRIBUTED_CACHE |
| 809 | bool use_distributed_cache = enable_distributed_cache |
| 810 | && DistributedCache::canUseDistributedCacheForRead( |
| 811 | read_settings, *storage); |
| 812 | #else |
| 813 | bool use_distributed_cache = false; |
| 814 | #endif |
| 815 | |
| 816 | const bool file_cache_enabled = storage->supportsCache() && read_settings.enable_filesystem_cache; |
| 817 | |
| 818 | /// Avoid cache fragmentation by choosing a bigger buffer size when filesystem cache is active. |
| 819 | /// Must be done before setSource, which stores read_settings in the pipeline. |
| 820 | bool prefer_bigger_buffer_size = read_settings.filesystem_cache_settings.prefer_bigger_buffer_size |
| 821 | && !read_settings.filesystem_cache_settings.read_if_exists_otherwise_bypass |
| 822 | && file_cache_enabled; |
| 823 | #if ENABLE_DISTRIBUTED_CACHE |
| 824 | if (use_distributed_cache && !read_settings.distributed_cache_settings.prefer_bigger_buffer_size) |
| 825 | prefer_bigger_buffer_size = false; |
| 826 | #endif |
| 827 | |
| 828 | if (prefer_bigger_buffer_size) |
| 829 | read_settings.remote_fs_settings.buffer_size = std::max<size_t>(read_settings.remote_fs_settings.buffer_size, read_settings.remote_fs_settings.large_buffer_size); |
| 830 | |
| 831 | /// Object storage files may be split across multiple blobs — gather joins them. |
| 832 | pipeline.needGather(); |
| 833 | |
| 834 | /// Delegate to the object storage to set source and add cache stage if needed. |
| 835 | /// CachedObjectStorage::prepareRead adds needFilesystemCache automatically. |
| 836 | storage->prepareRead(storage, storage_objects, read_settings, read_hint, pipeline); |
| 837 | |
| 838 | if (use_distributed_cache) |
| 839 | pipeline.needDistributedCache(); |
| 840 | |
| 841 | /// Memory cache (page cache). |
| 842 | const bool use_page_cache = |
| 843 | read_settings.page_cache_settings.cache |
| 844 | && (use_distributed_cache |