OIiveirra commented on code in PR #69528:
URL: https://github.com/apache/doris/pull/69528#discussion_r4367403255
##########
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:
Fixed in 2d7318284d162ad1dc9b1912a8e1150e1f74c674, with the same TVF-only
policy on the other active branches. HDFS TVF checks normalized routes before
listing and checks every returned path before schema/data use; FILE delegation
reaches the same checks. Java listing, native range NameNode and native schema
NameNode are checked using their separate consumer semantics.
Hadoop TVFs reject local/unknown delegation even when LOCAL is enabled,
because the generic Hadoop reader cannot enforce LOCAL privileges and
secure-root containment. The error directs local reads to constrained local();
generic Hadoop/catalog behavior is unchanged. Tests verify disabled/enabled
rejection before filesystem creation, actual normalizer/provider controls,
remote listing and returned-path checks. All 86 selected FE tests passed,
including the release Java trim and native schema-path controls; actual Hadoop
callbacks were recording-only.
##########
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:
Fixed in 2d7318284d162ad1dc9b1912a8e1150e1f74c674. Java TVF metadata loading
inspects manifest bases immediately after opening the primary snapshot and
before BASIC metadata use. BE TVFs use the new metadata-only Lance C API to
inspect SDK-decoded primary and every registered base before
schema/scanner/index/row-id work: remote TVFs reject local bases; LOCAL bases
must canonicalize inside the secure root and their reachable trees are checked.
Both dataset-root and direct-data-directory bases are covered. Policy is part
of the dataset key and is enforced on cached/reused handles; unmarked
fixed-version catalogs retain their configured policy.
All 86 selected FE and 22 selected BE tests passed with zero
failures/errors/skips. Official SDK-generated fixtures prove outside-root
rejection and actual in-root rows 1–4. Isolated S3 tests execute Java
rejection, BE remote/local-base rejection and actual remote rows 1–4. The C
ABI/SDK tests and matching header/archive checks passed. Complete enforcement
requires upgraded serving FE/BE versions; the PR describes the optional TVF
marker and mixed-version boundary.
--
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]
