This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 60818fd6716 [fix](filecache) Keep HTTP files out of the file cache and
key external files by storage (#68847)
60818fd6716 is described below
commit 60818fd6716c90f36ff434ced6b31fb4b42640fa
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 12 14:29:11 2026 +0800
[fix](filecache) Keep HTTP files out of the file cache and key external
files by storage (#68847)
### What problem does this PR solve?
Issue Number: None
Related PR: #68797, #66932
Problem Summary:
**In short.** The BE file cache can return the bytes of a different file
for two kinds of external reads. An `http()` URL whose content changes
keeps being served from blocks cached from the old content. Two object
storage endpoints, or two HDFS clusters, that hold the same path with
the same modification time share cache blocks. Both happen on master
whenever a session turns on `enable_file_cache`, and on branch-4.x,
where the cache is on by default. #68797 turns the cache on by default
on master and is blocked on these two cases. This PR keeps HTTP
responses out of the file cache and adds the storage to the cache key of
external files.
**Background.** `CachedRemoteFileReader` keys an external file as
`path:mtime`: the path its reader reports plus the modification time
from the file listing. The key is meant to change whenever the file
does. `FileFactory` wraps S3, HDFS and HTTP readers in the cache when
the query enables it. #66932 already exempted chunked HTTP responses
(the `cdc_stream` table function), whose content is a stream.
**The problem, and what it cost.**
- **HTTP has no version.** `HttpTableValuedFunction` lists a URL with
modification time 0, so every response of that URL gets the key `url:0`.
When the content behind the URL changes, a later query reads blocks
cached from the earlier content. If the length did not change, nothing
fails and the query silently returns the old rows; if it did, old and
new bytes can be mixed.
- **The key does not name the storage.** An S3 reader reports
`s3://bucket/key`, without the endpoint. An HDFS reader reports the path
with the name node stripped (`convert_path`). Two S3-compatible
endpoints (for example two MinIO deployments) holding the same bucket
and key, or two HDFS clusters holding the same path, get the same key as
soon as the modification times match. S3 reports them in seconds, so
files written by the same job in the same second match. A query on one
storage can then read blocks of the other storage's file.
**How this PR fixes it.**
- `FileFactory` no longer wraps HTTP readers in the file cache. This
covers every HTTP source read through `FileFactory`, including `hf://`
URLs, and replaces the chunked-response check from #66932, which this
rule includes.
- `FileReaderOptions` gets `cache_key_function`, which builds the cache
key of an external file from its path and modification time, so the code
that opens the file defines its key. `FileFactory` sets it for S3
readers to put the S3 endpoint (`AWS_ENDPOINT`) in front of the path,
and for HDFS readers the name node (`fs_name`): the key becomes
`<storage>:path:mtime`. `CachedRemoteFileReader` calls the function when
it is set; readers that do not set it, such as broker reads, keep
`path:mtime`.
**Results.**
| Case | Before | After |
|---|---|---|
| `http()` URL whose content was replaced, file cache on | second query
can return rows of the old content | always reads the current content;
HTTP data is not cached |
| Same bucket, key and mtime on two S3-compatible endpoints | one cache
entry, shared | one entry per endpoint |
| Same path and mtime on two HDFS clusters | one cache entry, shared |
one entry per name node |
| Existing cached blocks of external S3 / HDFS files after upgrade | - |
not found under the new key; the first reads go to the storage again and
the old blocks age out |
**Classes, and how they call each other.**
```
FileScanner / format readers
└─ FileFactory::create_file_reader(system_properties, file_description,
reader_options)
├─ FILE_S3 : options.cache_key_function = endpoint:path:mtime ─┐
├─ FILE_HDFS : options.cache_key_function = fs_name:path:mtime ──┤
│
create_cached_file_reader(reader, options)
│ └─
CachedRemoteFileReader::_init_external_table_cache
│ key =
cache_key_function(path, mtime)
│ (path:mtime
when it is not set)
└─ FILE_HTTP : HttpFileReader, not wrapped in the file cache
```
---
be/src/io/cache/cached_remote_file_reader.cpp | 7 ++-
be/src/io/file_factory.cpp | 42 +++++++++------
be/src/io/fs/file_reader.h | 5 ++
.../io/cache/cached_remote_file_reader_test.cpp | 60 ++++++++++++++++++++++
be/test/io/fs/http_file_reader_test.cpp | 19 +++++++
.../data/external_table_p0/tvf/test_http_tvf.out | 6 +++
.../external_table_p0/tvf/test_http_tvf.groovy | 32 +++++++++++-
7 files changed, 151 insertions(+), 20 deletions(-)
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp
b/be/src/io/cache/cached_remote_file_reader.cpp
index ef0c395b4c1..18eb7c3d00d 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -142,8 +142,11 @@ void CachedRemoteFileReader::_init_doris_table_cache() {
}
void CachedRemoteFileReader::_init_external_table_cache(const
FileReaderOptions& opts) {
- // Use path and modification time to build cache key.
- std::string unique_path = fmt::format("{}:{}", path().native(),
opts.mtime);
+ // Use path and modification time to build cache key, unless the creator
of this reader names
+ // the file by more than its path.
+ std::string unique_path = opts.cache_key_function
+ ?
opts.cache_key_function(path().native(), opts.mtime)
+ : fmt::format("{}:{}", path().native(),
opts.mtime);
_cache_hash = BlockFileCache::hash(unique_path);
if (opts.cache_base_path.empty()) {
// If cache path is not specified by session variable, choose randomly.
diff --git a/be/src/io/file_factory.cpp b/be/src/io/file_factory.cpp
index a29ac314545..c17742d9e2e 100644
--- a/be/src/io/file_factory.cpp
+++ b/be/src/io/file_factory.cpp
@@ -17,6 +17,7 @@
#include "io/file_factory.h"
+#include <fmt/format.h>
#include <gen_cpp/PaloInternalService_types.h>
#include <gen_cpp/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
@@ -52,13 +53,25 @@
#include "service/backend_options.h"
#include "util/s3_uri.h"
#include "util/s3_util.h"
-#include "util/string_util.h"
#include "util/uid_util.h"
namespace doris {
constexpr std::string_view RANDOM_CACHE_BASE_PATH = "random";
+// The file cache keys an external file by its path and modification time, but
a path names a file
+// only within one storage: the same bucket and key on two object storage
endpoints, or the same
+// path on two HDFS clusters, are different files. Put the storage in front of
the key.
+static io::FileReaderOptions with_storage_in_cache_key(const
io::FileReaderOptions& options,
+ std::string storage) {
+ auto keyed = options;
+ keyed.cache_key_function = [storage = std::move(storage)](const
std::string& path,
+ int64_t mtime) {
+ return fmt::format("{}:{}:{}", storage, path, mtime);
+ };
+ return keyed;
+}
+
io::FileReaderOptions FileFactory::get_reader_options(const TQueryOptions&
option,
const
io::FileDescription& fd) {
io::FileReaderOptions opts {
@@ -232,10 +245,11 @@ Result<io::FileReaderSPtr>
FileFactory::_create_file_reader_internal(
system_properties.properties, s3_uri, &s3_conf));
auto client_holder =
std::make_shared<io::ObjClientHolder>(s3_conf.client_conf);
RETURN_IF_ERROR_RESULT(client_holder->init());
+ auto options = with_storage_in_cache_key(reader_options,
s3_conf.client_conf.endpoint);
return io::S3FileReader::create(std::move(client_holder),
s3_conf.bucket, s3_uri.get_key(),
file_description.file_size, profile)
- .and_then([&](auto&& reader) {
- return io::create_cached_file_reader(std::move(reader),
reader_options);
+ .and_then([&options](auto&& reader) {
+ return io::create_cached_file_reader(std::move(reader),
options);
});
}
case TFileType::FILE_HDFS: {
@@ -248,10 +262,12 @@ Result<io::FileReaderSPtr>
FileFactory::_create_file_reader_internal(
}
RETURN_IF_ERROR_RESULT(ExecEnv::GetInstance()->hdfs_mgr()->get_or_create_fs(
system_properties.hdfs_params, *fs_name, &handler));
+ // The reader's path drops the name node, so the cache key takes it
from here.
+ auto options = with_storage_in_cache_key(reader_options, *fs_name);
return io::HdfsFileReader::create(file_description.path,
handler->hdfs_fs, *fs_name,
- reader_options)
- .and_then([&](auto&& reader) {
- return io::create_cached_file_reader(std::move(reader),
reader_options);
+ options)
+ .and_then([&options](auto&& reader) {
+ return io::create_cached_file_reader(std::move(reader),
options);
});
}
case TFileType::FILE_BROKER: {
@@ -272,17 +288,11 @@ Result<io::FileReaderSPtr>
FileFactory::_create_file_reader_internal(
});
}
case TFileType::FILE_HTTP: {
- auto options = reader_options;
- auto chunk_response =
system_properties.properties.find("http.enable.chunk.response");
- if (chunk_response != system_properties.properties.end() &&
- (iequal(chunk_response->second, "true") || chunk_response->second
== "1")) {
- options.cache_type = io::FileCachePolicy::NO_CACHE;
- }
+ // HTTP responses never enter the file cache. The cache key names an
external file by its
+ // path and modification time, but an HTTP source has no modification
time, so a URL whose
+ // content changes would keep serving the blocks cached from the old
content.
return io::HttpFileReader::create(file_description.path,
system_properties.properties,
- options, profile)
- .and_then([&options](auto&& reader) {
- return io::create_cached_file_reader(std::move(reader),
options);
- });
+ reader_options, profile);
}
default:
return ResultError(
diff --git a/be/src/io/fs/file_reader.h b/be/src/io/fs/file_reader.h
index 6633841dc45..bc736c4a764 100644
--- a/be/src/io/fs/file_reader.h
+++ b/be/src/io/fs/file_reader.h
@@ -20,6 +20,7 @@
#include <butil/macros.h>
#include <stddef.h>
+#include <functional>
#include <memory>
#include <string>
@@ -81,6 +82,10 @@ struct FileReaderOptions {
// Storage resource id of the remote file system. Used by peer fill to
reconstruct
// the source file system without scanning tablet rowsets on the peer.
std::string storage_resource_id;
+ // Builds the file cache key of an external file from its path and
modification time. The
+ // creator of the reader sets it when the path alone does not name the
file, e.g. FileFactory
+ // adds the object storage endpoint or the HDFS name node. When it is
empty the key is path:mtime.
+ std::function<std::string(const std::string& path, int64_t mtime)>
cache_key_function {};
static const FileReaderOptions DEFAULT;
};
diff --git a/be/test/io/cache/cached_remote_file_reader_test.cpp
b/be/test/io/cache/cached_remote_file_reader_test.cpp
index 6fe9bac5d60..98edb308a7b 100644
--- a/be/test/io/cache/cached_remote_file_reader_test.cpp
+++ b/be/test/io/cache/cached_remote_file_reader_test.cpp
@@ -22,6 +22,7 @@
#include "cloud/config.h"
#include "io/cache/async_cache_write_manager_metrics.h"
#include "io/cache/inflight_write_buffer_index.h"
+#include "io/file_factory.h"
#include "util/time.h"
namespace doris::io {
@@ -228,6 +229,65 @@ TEST_F(AsyncCachedRemoteFileReaderTest,
external_reader_normalizes_tablet_id_for
}
}
+TEST_F(AsyncCachedRemoteFileReaderTest,
external_cache_key_comes_from_the_key_function) {
+ create_cache("cached_external_reader_key_function");
+ constexpr int64_t mtime = 1700000000000;
+ auto external_reader =
+ [this](std::function<std::string(const std::string&, int64_t)>
cache_key_function) {
+ FileReaderOptions options;
+ options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+ options.mtime = mtime;
+ options.cache_key_function = std::move(cache_key_function);
+ return
std::make_shared<CachedRemoteFileReader>(open_remote_file(), options);
+ };
+
+ // The reader keys the file by what the key function makes of its path and
modification time.
+ auto keyed = external_reader([](const std::string& path, int64_t
file_mtime) {
+ return fmt::format("http://storage-a:9000:{}:{}";, path, file_mtime);
+ });
+ EXPECT_EQ(keyed->_cache_hash,
BlockFileCache::hash(fmt::format("http://storage-a:9000:{}:{}";,
+
keyed->path().native(), mtime)));
+
+ // Without a key function the key is path:mtime.
+ auto unkeyed = external_reader(nullptr);
+ EXPECT_EQ(unkeyed->_cache_hash,
+ BlockFileCache::hash(fmt::format("{}:{}",
unkeyed->path().native(), mtime)));
+}
+
+TEST_F(AsyncCachedRemoteFileReaderTest,
file_factory_keys_s3_files_by_endpoint) {
+ create_cache("file_factory_s3_endpoint_key");
+ constexpr int64_t mtime = 1700000000000;
+ auto s3_reader = [](const std::string& endpoint) ->
std::shared_ptr<CachedRemoteFileReader> {
+ FileSystemProperties properties;
+ properties.system_type = TFileType::FILE_S3;
+ properties.properties = {{"AWS_ENDPOINT", endpoint},
+ {"AWS_REGION", "us-east-1"},
+ {"AWS_ACCESS_KEY", "ak"},
+ {"AWS_SECRET_KEY", "sk"}};
+ FileDescription file_description;
+ file_description.path = "s3://bucket/data/file.parquet";
+ // A known size skips the HEAD request, so no object storage is needed.
+ file_description.file_size = 128;
+ file_description.mtime = mtime;
+ FileReaderOptions options;
+ options.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+ options.mtime = mtime;
+ auto reader = FileFactory::create_file_reader(properties,
file_description, options);
+ EXPECT_TRUE(reader.has_value()) << reader.error();
+ return
std::dynamic_pointer_cast<CachedRemoteFileReader>(reader.value());
+ };
+
+ // Two endpoints serving the same bucket, key and modification time do not
share cache blocks.
+ auto endpoint_a = s3_reader("http://127.0.0.1:9000";);
+ auto endpoint_b = s3_reader("http://127.0.0.2:9000";);
+ ASSERT_NE(endpoint_a, nullptr);
+ ASSERT_NE(endpoint_b, nullptr);
+ EXPECT_NE(endpoint_a->_cache_hash, endpoint_b->_cache_hash);
+ EXPECT_EQ(endpoint_a->_cache_hash,
+ BlockFileCache::hash(fmt::format(
+
"http://127.0.0.1:9000:s3://bucket/data/file.parquet:{}";, mtime)));
+}
+
TEST_F(AsyncCachedRemoteFileReaderTest,
preallocated_cache_block_can_cover_the_short_file_tail) {
create_cache("cached_remote_reader_async_preallocated_file_tail");
auto counting_reader =
std::make_shared<CountingFileReader>(open_remote_file());
diff --git a/be/test/io/fs/http_file_reader_test.cpp
b/be/test/io/fs/http_file_reader_test.cpp
index 1c63ce3fd21..d55ff5fa888 100644
--- a/be/test/io/fs/http_file_reader_test.cpp
+++ b/be/test/io/fs/http_file_reader_test.cpp
@@ -97,6 +97,25 @@ TEST(HttpFileReaderTest, ChunkResponseDisablesFileCache) {
EXPECT_NE(std::dynamic_pointer_cast<HttpFileReader>(reader.value()),
nullptr);
}
+TEST(HttpFileReaderTest, PlainResponseIsNotFileCachedEither) {
+ // The file cache keys an external file by path and modification time. An
HTTP source has no
+ // modification time, so a URL whose content changes must not be served
from blocks cached
+ // from its old content.
+ FileSystemProperties properties;
+ properties.system_type = TFileType::FILE_HTTP;
+ // A known size skips the HEAD request, so no server is needed.
+ properties.properties = {{"file_size", "4"}};
+ FileDescription file_description;
+ file_description.path = "http://127.0.0.1/data.csv";;
+ FileReaderOptions opts;
+ opts.cache_type = FileCachePolicy::FILE_BLOCK_CACHE;
+
+ auto reader = FileFactory::create_file_reader(properties,
file_description, opts);
+
+ ASSERT_TRUE(reader.has_value()) << reader.error();
+ EXPECT_NE(std::dynamic_pointer_cast<HttpFileReader>(reader.value()),
nullptr);
+}
+
TEST(HttpFileReaderTest, RangeReadDoesNotCrossKnownEof) {
const std::string content =
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
EofSensitiveRangeHandler handler(content);
diff --git a/regression-test/data/external_table_p0/tvf/test_http_tvf.out
b/regression-test/data/external_table_p0/tvf/test_http_tvf.out
index 6cc959956bd..a097aafa9f3 100644
--- a/regression-test/data/external_table_p0/tvf/test_http_tvf.out
+++ b/regression-test/data/external_table_p0/tvf/test_http_tvf.out
@@ -159,6 +159,12 @@ m map<text,text> Yes false \N NONE
-- !sql16 --
204
+-- !replaced_before --
+1 aaaa
+
+-- !replaced_after --
+2 bbbb
+
-- !sql17 --
c1 text Yes false \N NONE
diff --git a/regression-test/suites/external_table_p0/tvf/test_http_tvf.groovy
b/regression-test/suites/external_table_p0/tvf/test_http_tvf.groovy
index f22eb702a8e..485d267e48a 100644
--- a/regression-test/suites/external_table_p0/tvf/test_http_tvf.groovy
+++ b/regression-test/suites/external_table_p0/tvf/test_http_tvf.groovy
@@ -27,6 +27,8 @@ import java.net.URLDecoder
suite("test_http_tvf", "p0") {
def dataRoot =
Paths.get(context.config.dataPath).toAbsolutePath().normalize()
+ // Served under mutable/: files whose content a case replaces between two
queries.
+ def mutableRoot =
Files.createTempDirectory("test_http_tvf_mutable").toAbsolutePath().normalize()
HttpServer httpServer = HttpServer.create(new InetSocketAddress("0.0.0.0",
0), 0)
def writeResponse = { HttpExchange exchange, int status, byte[] body ->
@@ -77,8 +79,13 @@ suite("test_http_tvf", "p0") {
def requestPath = URLDecoder.decode(exchange.requestURI.path, "UTF-8")
def relativePath = requestPath.startsWith("/") ?
requestPath.substring(1) : requestPath
- def filePath = dataRoot.resolve(relativePath).normalize()
- if (!filePath.startsWith(dataRoot)) {
+ def root = dataRoot
+ if (relativePath.startsWith("mutable/")) {
+ root = mutableRoot
+ relativePath = relativePath.substring("mutable/".length())
+ }
+ def filePath = root.resolve(relativePath).normalize()
+ if (!filePath.startsWith(root)) {
writeResponse(exchange, 403, "Forbidden".getBytes("UTF-8"))
return
}
@@ -321,6 +328,27 @@ suite("test_http_tvf", "p0") {
) order by id;
"""
+ // The same URL serving new content of the same length: an HTTP response
has no modification
+ // time for the file cache key, so the second query must not read blocks
cached by the first.
+ sql """ set enable_file_cache = true """
+ def mutableCsv = mutableRoot.resolve("replaced.csv")
+ Files.write(mutableCsv, "1,aaaa\n".getBytes("UTF-8"))
+ order_qt_replaced_before """
+ select * from http(
+ "uri" = "${httpUrl("mutable/replaced.csv")}",
+ "format" = "csv",
+ "column_separator" = ","
+ );
+ """
+ Files.write(mutableCsv, "2,bbbb\n".getBytes("UTF-8"))
+ order_qt_replaced_after """
+ select * from http(
+ "uri" = "${httpUrl("mutable/replaced.csv")}",
+ "format" = "csv",
+ "column_separator" = ","
+ );
+ """
+
// hf
def hfCountOneFile = sql """
select count(*) from
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
