github-actions[bot] commented on code in PR #69528:
URL: https://github.com/apache/doris/pull/69528#discussion_r4366880442
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -22,24 +22,150 @@
#include <arrow/record_batch.h>
#include <arrow/type.h>
#include <lance/lance.h>
+#include <pwd.h>
+#include <unistd.h>
#include <algorithm>
#include <bit>
+#include <cerrno>
+#include <cstdlib>
#include <cstring>
+#include <filesystem>
#include <limits>
#include <memory>
+#include <unordered_set>
#include "common/config.h"
#include "common/logging.h"
#include "exec/common/endian.h"
#include "format_v2/lance/lance_reader_helper.h"
#include "format_v2/lance/lance_runtime_filter_helper.h"
#include "format_v2/lance/lance_session_manager.h"
+#include "io/fs/local_file_system.h"
#include "runtime/file_scan_profile.h"
#include "runtime/runtime_state.h"
namespace doris::format::lance {
+namespace {
+std::string expand_home_path(const std::string& path) {
+ if (path != "~" && !path.starts_with("~/")) {
+ return path;
+ }
+ // Match the SDK's std::env::home_dir: HOME first, then the current user's
passwd entry.
+ std::string home;
+ if (const char* value = std::getenv("HOME"); value != nullptr && value[0]
!= '\0') {
+ home = value;
+ } else {
+ std::vector<char> buffer(1024);
+ struct passwd entry {};
+ struct passwd* result = nullptr;
+ int error;
+ while ((error = getpwuid_r(getuid(), &entry, buffer.data(),
buffer.size(), &result)) ==
+ ERANGE) {
+ buffer.resize(buffer.size() * 2);
+ }
+ if (error == 0 && result != nullptr && entry.pw_dir != nullptr) {
+ home = entry.pw_dir;
+ }
+ }
+ if (home.empty()) {
+ return path;
+ }
+ return path == "~" ? home : (io::Path(home) / path.substr(2)).string();
+}
+
+Status validate_local_dataset_files(const std::string& dataset_uri) {
+ // Lance opens manifests and fragments itself, so a confined dataset
directory is not
+ // enough: its descendants can redirect those opens through symlinks.
+ std::error_code error;
+ const auto secure_root =
std::filesystem::canonical(config::user_files_secure_path, error);
+ if (error) {
+ return Status::InvalidArgument("local Lance dataset path is not
allowed");
+ }
+ std::vector<std::filesystem::path> pending {dataset_uri};
+ std::unordered_set<std::string> visited;
+ while (!pending.empty()) {
+ auto directory = std::move(pending.back());
+ pending.pop_back();
+ // Canonical directory identity bounds cycles and repeated in-root
aliases without
+ // imposing a file-count or nesting limit on otherwise valid datasets.
+ if (!visited.insert(directory.string()).second) {
+ continue;
+ }
+ std::filesystem::directory_iterator entries(directory, error);
+ if (error) {
+ return Status::InvalidArgument("local Lance dataset path is not
allowed");
+ }
+ const std::filesystem::directory_iterator end;
+ while (entries != end) {
+ auto target = entries->path();
+ auto status = entries->symlink_status(error);
+ if (error) {
+ return Status::InvalidArgument("local Lance dataset path is
not allowed");
+ }
+ if (std::filesystem::is_symlink(status)) {
+ target = std::filesystem::canonical(target, error);
+ if (error || !io::LocalFileSystem::contain_path(secure_root,
target)) {
+ return Status::InvalidArgument("local Lance dataset path
is not allowed");
+ }
+ status = std::filesystem::status(target, error);
+ if (error) {
+ return Status::InvalidArgument("local Lance dataset path
is not allowed");
+ }
+ }
+ if (std::filesystem::is_directory(status)) {
+ pending.push_back(std::move(target));
+ }
+ entries.increment(error);
+ if (error) {
+ return Status::InvalidArgument("local Lance dataset path is
not allowed");
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status resolve_dataset_uri(const TFileRangeDesc& range, const
TFileScanRangeParams& scan_params,
+ std::string* uri) {
+ const auto& requested_uri =
range.table_format_params.lance_params.dataset_uri;
+ // Modern plans tag each range; older plans put the type in the scan
parameters.
+ // Untagged catalog scans retain their original storage-provider selection.
+ const bool is_local = range.__isset.file_type
+ ? range.file_type == TFileType::FILE_LOCAL
+ : scan_params.__isset.file_type &&
+ scan_params.file_type ==
TFileType::FILE_LOCAL;
+ // LOCAL TVF uses version zero for both schema discovery and whole-dataset
scans.
+ // Catalog scans pin a positive snapshot version, including file://
catalogs whose
+ // operator-configured warehouse is independent of the LOCAL TVF secure
directory.
+ if (is_local && range.table_format_params.lance_params.version == 0) {
+ // LOCAL Lance bypasses glob because it opens a dataset directory.
Apply the same
+ // canonical secure-root check before either schema discovery or a
cached dataset open.
+ std::string local_input = requested_uri;
+ const auto colon = requested_uri.find(':');
+ const auto slash = requested_uri.find('/');
+ const bool bare_path =
+ colon == std::string::npos || (slash != std::string::npos &&
slash < colon);
+ if (bare_path && !requested_uri.empty() && requested_uri.find('\0') ==
std::string::npos &&
+ !config::user_files_secure_path.empty()) {
+ // Lance's PathAbs fallback resolves bare dot segments before
following symlinks.
+ // Anchor relative TVF paths to the canonical secure root, not the
process cwd.
+ std::string secure_root;
+ RETURN_IF_ERROR(io::global_local_filesystem()->canonicalize(
+ config::user_files_secure_path, &secure_root));
+ local_input = (io::Path(secure_root) /
expand_home_path(requested_uri))
+ .lexically_normal()
+ .string();
+ }
+ RETURN_IF_ERROR(io::global_local_filesystem()->canonicalize_local_file(
+ config::user_files_secure_path, local_input, uri));
+ return validate_local_dataset_files(*uri);
Review Comment:
[P1] Validate manifest-selected storage before Lance opens fragments
This walk checks physical descendants, but valid Lance manifests can select
another storage base without symlinks. The [pinned SDK's shallow
clone](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-table/src/format/manifest.rs#L241-L305)
assigns `DataFile.base_id` and records the source URI. With LOCAL enabled,
`/secure/clone.lance` can contain only regular metadata yet reference
`/private/source.lance`: schema discovery and this walk succeed, then the SDK
[resolves the external
base/provider](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance/src/dataset.rs#L2441-L2493)
and [opens its
fragments](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance/src/dataset/fragment.rs#L1420-L1454)
outside `user_files_secure_path`.
The same mechanism also bypasses the default-off switch: an ordinary S3 TVF
caller can supply a valid `s3://bucket/clone.lance` manifest with a compatible
scalar schema, known readable local fragment, and `base_id ->
file:///private/source.lance`. `LanceMetadataLoader` BASIC loading only reads
version/schema/fragment fields, and the top-level remote-URI guard accepts it;
the `FILE_S3` fixed-version BE range skips this local branch and opens that
local base. This needs no malformed URI and is distinct from the existing
parser threads. Validate effective manifest bases before TVF data opens: reject
local bases for remote TVFs and enforce canonical containment for LOCAL bases,
or reject external-base TVFs until that boundary is enforceable. Cover
out-of-root and entirely in-root shallow clones.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java:
##########
@@ -135,6 +150,332 @@ private void getFileListFromBackend() throws
AnalysisException {
}
}
+ /** Apply LOCAL admission to the effective sink before planning or
filesystem side effects. */
+ public static void checkSinkEnabled(String tvfName, Map<String, String>
properties) {
+ if ("local".equalsIgnoreCase(tvfName) ||
isLocalWriter(properties.getOrDefault("writer_class", ""))) {
+ checkEnabled();
+ }
+ if ("hdfs".equalsIgnoreCase(tvfName)) {
+ if (!properties.containsKey("writer_class")) {
+ checkHdfsSinkEnabled(properties);
+ }
+ if
(Boolean.parseBoolean(properties.getOrDefault("delete_existing_files",
"false"))) {
+ checkHadoopFileSystemEnabled(
+
BrokerUtil.extractParentDirectory(properties.getOrDefault("file_path", "")),
properties);
+ }
+ }
+ }
+
+ private static boolean isLocalWriter(String writer) {
+ // ScannerLoader accepts slashes, and JNI passes the class name as a C
string.
+ return
"org.apache.doris.writer.LocalFileJniWriter".equals(nativeStringPrefix(writer).replace('/',
'.'));
+ }
+
+ /** Check the NameNode and provider used by the native FILE_HDFS writer. */
+ public static void checkHdfsSinkEnabled(Map<String, String> properties) {
+ checkHdfsSinkEnabled(properties.getOrDefault("file_path", ""),
properties);
+ }
+
+ public static void checkHdfsSinkEnabled(String outputPrefix, Map<String,
String> properties) {
+ String nameNode =
nativeStringPrefix(properties.getOrDefault("fs.defaultFS", "")).trim();
+ if ("default".equals(nameNode)) {
+ // libhdfs delegates this selector to the BE's defaults, which may
select local storage.
+ checkEnabled();
+ }
+ // libhdfs prefixes bare NameNodes with hdfs:// rather than using the
output path's scheme.
+ checkHadoopSchemeEnabled(nameNode.contains("://") ?
uriScheme(nameNode) : "hdfs", properties);
+ if (!Config.enable_local_tvf) {
+ try {
+ URI filesystem = new URI(nameNode.contains("://") ? nameNode :
"hdfs://" + nameNode);
+ // BE appends a generated filename directly to this prefix,
then convert_path
+ // removes the NameNode. Resolve its parent, not an exact
mount-point filename.
+ String generated = outputPrefix + "doris-tvf-file";
+ String rawNameNode = properties.getOrDefault("fs.defaultFS",
"");
+ if (!rawNameNode.isEmpty() && generated.contains(rawNameNode))
{
+ generated = generated.substring(rawNameNode.length());
+ }
+ if (!generated.startsWith("/")) {
+ generated = "/" + generated;
+ }
+ String path =
hadoopPathUri(nativeStringPrefix(generated)).getPath();
+ path = path.substring(0, path.lastIndexOf('/') + 1);
+ if (path.length() > 1) {
+ path = path.substring(0, path.length() - 1);
+ }
+ checkViewFsEnabled(filesystem, path, properties, new
HashSet<>());
+ } catch (URISyntaxException e) {
+ // Invalid NameNodes remain subject to the existing storage
validation.
+ }
+ }
+ }
+
+ /** Check Hadoop's path/default-filesystem selection before FE-side
cleanup. */
+ public static void checkHadoopFileSystemEnabled(String path, Map<String,
String> properties) {
Review Comment:
[P1] Apply effective-filesystem admission to Hadoop TVF reads
With `enable_local_tvf=false`, an ordinary user can still execute `SELECT *
FROM file("uri"="viewfs://review/local/data.csv",
"fs.defaultFS"="viewfs://review",
"fs.viewfs.mounttable.review.link./local"="file:///tmp", "format"="csv",
"csv_schema"="id:int")`. Given a readable `/tmp/data.csv` on the FE and
selected BE, this reads the BE-local file even when `/tmp` is outside
`user_files_secure_path`. FILE delegates to `HdfsTableValuedFunction`; its
`init -> parseFile -> DFSFileSystem.globList` selects ViewFS without calling
this admission helper or LOCAL's ADMIN/OPERATOR check. The range stays
`FILE_HDFS`, and `FileFactory -> HdfsFileReader` reaches the [pinned ViewFS
target
open](https://github.com/apache/doris-thirdparty/blob/hadoop-3.4.2.2-for-doris/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java#L686-L690),
bypassing BE local containment too. `hdfs(...)` exposes the same read route.
This is distinct from the existing native INSERT
/cleanup thread: its fixes only invoke these helpers on writes/deletion. Check
the effective normalized filesystem before Hadoop TVF listing, including FILE
delegation, and enforce LOCAL read privileges/containment for local targets;
add disabled-switch and truly remote read controls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
