[BugFix] Check read size before reading data block from datacache. (backport #38722) (#38799)

Signed-off-by: Gavin <[email protected]>

Author: GavinMar

be/src/io/cache_input_stream.cpp

@@ -54,8 +54,11 @@ Status CacheInputStream::read_at_fully(int64_t offset, void* out, int64_t count)
     char* pe = p + count;
 
     auto read_one_block = [&](size_t offset, size_t size) {
-        StatusOr<size_t> res;
+        if (UNLIKELY(size == 0)) {
+            return Status::OK();
+        }
 
+        StatusOr<size_t> res;
         DCHECK(size <= BLOCK_SIZE);
         {
             SCOPED_RAW_TIMER(&_stats.read_cache_ns);

be/test/io/cache_input_stream_test.cpp

@@ -189,4 +189,27 @@ TEST_F(CacheInputStreamTest, test_file_overwrite) {
     ASSERT_EQ(stats2.read_cache_count, 0);
 }
 
+TEST_F(CacheInputStreamTest, test_read_with_zero_range) {
+    const int64_t block_count = 1;
+    int64_t data_size = block_size * block_count;
+    char data[data_size + 1];
+    gen_test_data(data, data_size, block_size);
+
+    std::shared_ptr<io::SeekableInputStream> stream(new MockSeekableInputStream(data, data_size));
+    io::CacheInputStream cache_stream(stream, "test_file4", data_size, 1000000);
+    cache_stream.set_enable_populate_cache(true);
+    auto& stats = cache_stream.stats();
+
+    // read from backend, cache the data
+    char buffer[block_size];
+    read_stream_data(&cache_stream, 0, block_size, buffer);
+    ASSERT_TRUE(check_data_content(buffer, block_size, 'a'));
+    ASSERT_EQ(stats.read_cache_count, 0);
+    ASSERT_EQ(stats.write_cache_count, 1);
+
+    // try read zero length data, expect no crash
+    read_stream_data(&cache_stream, 0, 0, nullptr);
+    ASSERT_EQ(stats.read_cache_count, 0);
+}
+
 } // namespace starrocks::io