OIiveirra commented on code in PR #69528:
URL: https://github.com/apache/doris/pull/69528#discussion_r4369659423
##########
fe/be-java-extensions/hadoop-deps/src/main/java/org/apache/hadoop/fs/FileSystem.java:
##########
@@ -3622,13 +3633,628 @@ public static Class<? extends FileSystem>
getFileSystemClass(String scheme,
* @return the initialized filesystem.
* @throws IOException problems loading or initializing the FileSystem
*/
+ // DORIS-PATCH: TVF policy is checked against the frozen configuration used
by the consumer.
+ public static int getDorisTvfGuardVersion() {
+ return 1;
+ }
+
+ public static final class TvfGuard {
+ public static final String POLICY_KEY = "fs.doris.tvf.remote.paths";
+ public static final String NATIVE_POLICY = "native";
+ public static final String JAVA_READ_POLICY = "java-read";
+ public static final String ROOT_KEY = "fs.doris.tvf.root";
+ public static final String PATH_KEY = "fs.doris.tvf.path";
+ public static final String CONTEXT_KEY = "fs.doris.tvf.context";
+ private static final ThreadLocal<Context> LAST_NATIVE_CONTEXT = new
ThreadLocal<>();
+
+ private TvfGuard() {
+ }
+
+ public static final class Rule {
+ private final String filesystem;
+ private final String path;
+ private final boolean descendants;
+
+ public Rule(String filesystem, String path, boolean descendants) {
+ this.filesystem = filesystem;
+ this.path = path;
+ this.descendants = descendants;
+ }
+ }
+
+ public static String encode(List<Rule> rules) {
+ if (FileSystem.getDorisTvfGuardVersion() != 1) {
+ throw new IllegalStateException("TVF Hadoop runtime policy version
mismatch");
+ }
+ try {
+ java.io.ByteArrayOutputStream bytes = new
java.io.ByteArrayOutputStream();
+ java.io.DataOutputStream out = new java.io.DataOutputStream(bytes);
+ out.writeInt(1);
+ out.writeInt(rules.size());
+ for (Rule rule : rules) {
+ out.writeUTF(rule.filesystem);
+ out.writeUTF(rule.path);
+ out.writeBoolean(rule.descendants);
+ }
+ out.flush();
+ return
java.util.Base64.getEncoder().encodeToString(bytes.toByteArray());
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot encode TVF Hadoop route
policy", e);
+ }
+ }
+
+ private static IOException rejected() {
+ return new IOException("Hadoop TVF requires a proven remote filesystem
route. "
+ + "Use local() with LOCAL TVF privileges and secure-path
restrictions.");
+ }
+
+ private static Configuration freeze(Configuration conf) throws IOException
{
+ // Service providers (notably HDFS) register their default/site
resources on first load.
+ // Load them before copying raw values, as an ordinary FileSystem lookup
does.
+ if (!FILE_SYSTEMS_LOADED) {
+ loadFileSystems();
+ }
+ if (conf instanceof FrozenConfiguration) {
+ return conf;
+ }
+ return new FrozenConfiguration(conf);
+ }
+
+ private static Configuration prepare(Configuration conf) throws
IOException {
+ String policy = conf.getRaw(POLICY_KEY);
+ if (policy == null || policy.isEmpty()) {
+ return conf;
+ }
+ Configuration frozen = freeze(conf);
+ Context context = ((FrozenConfiguration) frozen).context;
+ if (NATIVE_POLICY.equals(policy)) {
+ context.checkNative(frozen.getRaw(ROOT_KEY), frozen.getRaw(PATH_KEY));
+ } else if (JAVA_READ_POLICY.equals(policy)) {
+ context.checkJava(new Path(frozen.getRaw(PATH_KEY)));
+ } else {
+ context.checkRules(policy);
+ }
+ return frozen;
+ }
+
+ public static void check(URI ignored, Configuration conf) throws
IOException {
+ prepare(conf);
+ }
+
+ public static void clearNativeContext() {
+ LAST_NATIVE_CONTEXT.remove();
+ }
+
+ public static Context takeNativeContext() {
+ Context result = LAST_NATIVE_CONTEXT.get();
+ LAST_NATIVE_CONTEXT.remove();
+ return result;
+ }
+
+ public static Configuration prepareJavaRead(Configuration conf, Path path,
String marker) throws IOException {
+ if (marker == null || marker.isEmpty()) {
+ return conf;
+ }
+ if (!NATIVE_POLICY.equals(marker)) {
+ throw rejected();
+ }
+ Configuration request = new Configuration(conf);
+ request.set(POLICY_KEY, JAVA_READ_POLICY);
+ request.set(ROOT_KEY, path.toUri().toString());
+ request.set(PATH_KEY, path.toString());
+ return prepare(request);
+ }
+
+ public static void checkJavaRead(FileSystem fs, Path path, String marker)
throws IOException {
+ if (marker == null || marker.isEmpty()) {
+ return;
+ }
+ if (!NATIVE_POLICY.equals(marker)) {
+ throw rejected();
+ }
+ // The marker is request-owned, never inferred from an unmarked cached
filesystem.
+ Configuration conf = fs.getConf();
+ if (JAVA_READ_POLICY.equals(conf.getRaw(POLICY_KEY))) {
+ // S3A derives a Configuration for bucket options. Its copy preserves
our internal
+ // marker and raw snapshot values, but not the private Configuration
subclass.
+ ((FrozenConfiguration) freeze(conf)).context.checkJava(path);
+ } else {
+ throw rejected();
+ }
+ }
+
+ private static final class FrozenConfiguration extends Configuration {
+ private final Context context;
+
+ private FrozenConfiguration(Configuration source) throws IOException {
+ super(false);
+ setClassLoader(source.getClassLoader());
+ java.util.SortedMap<String, String> values = new java.util.TreeMap<>();
+ for (Map.Entry<String, String> entry : source) {
+ String key = entry.getKey();
+ String value = source.getRaw(key);
+ set(key, value);
+ if (!PATH_KEY.equals(key) && !ROOT_KEY.equals(key) &&
!CONTEXT_KEY.equals(key)) {
+ values.put(key, value);
+ }
+ }
+ try {
+ java.security.MessageDigest digest =
java.security.MessageDigest.getInstance("SHA-256");
+ java.io.DataOutputStream output = new java.io.DataOutputStream(
+ new java.security.DigestOutputStream(new java.io.OutputStream() {
+ @Override
+ public void write(int value) {
+ }
+
+ @Override
+ public void write(byte[] values, int offset, int length) {
+ }
+ }, digest));
+ for (Map.Entry<String, String> entry : values.entrySet()) {
+ for (String value : new String[] {entry.getKey(),
entry.getValue()}) {
+ // Preserve raw Java code units, including unpaired surrogates.
UTF-8 replacement
+ // would give a literal '?' and an unpaired surrogate the same
cache profile.
+ output.writeInt(value.length());
+ output.writeChars(value);
+ }
+ }
+ output.flush();
+ StringBuilder identity = new StringBuilder();
+ for (byte value : digest.digest()) {
+ identity.append(Character.forDigit((value & 255) >>> 4, 16));
+ identity.append(Character.forDigit(value & 15, 16));
+ }
+ set(CONTEXT_KEY, identity.toString());
+ } catch (java.security.NoSuchAlgorithmException e) {
+ throw new IOException("SHA-256 is required for TVF Hadoop cache
isolation");
+ }
+ context = new Context(this);
+ }
+ }
+
+ public static final class Context {
+ private final Configuration conf;
+ private final Map<String, String> routingKeys = new HashMap<>();
+
+ private Context(Configuration conf) {
+ this.conf = conf;
+ for (Map.Entry<String, String> entry : conf) {
+ String key = entry.getKey();
+ if ("fs.defaultFS".equals(key) || "fs.default.name".equals(key)
+ || key.startsWith("fs.viewfs.") || (key.startsWith("fs.") &&
key.endsWith(".impl"))) {
+ routingKeys.put(key, entry.getValue());
+ }
+ }
+ }
+
+ private EffectiveRoute route(boolean descendants) {
+ return new EffectiveRoute(routingKeys, descendants, conf);
+ }
+
+ public String profile() {
+ return conf.getRaw(CONTEXT_KEY);
+ }
+
+ public boolean isDirectRemote() throws IOException {
+ String root = conf.getRaw(ROOT_KEY);
+ if (root == null) {
+ throw rejected();
+ }
+ try {
+ URI uri = nativeUri(root);
+ return route(false).provider(uri.getScheme()) ==
Route.Provider.REMOTE;
+ } catch (RuntimeException | URISyntaxException e) {
+ throw rejected();
+ }
+ }
+
+ private static URI nativeUri(String root) throws URISyntaxException {
+ root = root.trim();
+ if (root.isEmpty() || "default".equals(root)) {
+ throw new URISyntaxException("", "Unproven native filesystem");
+ }
+ return new URI(root.contains("://") ? root : "hdfs://" + root);
+ }
+
+ public void checkNative(String root, String path) throws IOException {
+ if (root == null || path == null || path.isEmpty()) {
+ throw rejected();
+ }
+ try {
+ URI uri = nativeUri(root);
+ String actual = new Path(path).toUri().getPath();
+ route(false).checkRoute(uri, actual, new HashSet<>());
+ } catch (RuntimeException | URISyntaxException e) {
+ throw rejected();
+ }
+ }
+
+ private void checkJava(Path path) throws IOException {
+ try {
+ URI uri = path.toUri();
+ route(false).checkRoute(uri, uri.getPath(), new HashSet<>());
+ } catch (RuntimeException | URISyntaxException e) {
+ throw rejected();
+ }
+ }
+
+ private void checkRules(String policy) throws IOException {
+ try {
+ byte[] bytes = java.util.Base64.getDecoder().decode(policy);
+ java.io.DataInputStream in = new java.io.DataInputStream(new
java.io.ByteArrayInputStream(bytes));
+ int version = in.readInt();
+ int count = in.readInt();
+ if (version != 1 || count <= 0 || count > in.available() / 5) {
+ throw rejected();
+ }
+ for (int i = 0; i < count; i++) {
+ URI filesystem = new URI(in.readUTF());
+ String path = in.readUTF();
+ boolean descendants = in.readBoolean();
+ route(descendants).checkRoute(filesystem, path, new HashSet<>());
+ }
+ if (in.available() != 0) {
+ throw rejected();
+ }
+ } catch (RuntimeException | URISyntaxException e) {
+ throw rejected();
+ }
+ }
+ }
+
+ private static final class EffectiveRoute extends Route {
+ private final Configuration conf;
+
+ private EffectiveRoute(Map<String, String> properties, boolean
descendants, Configuration conf) {
+ super(properties, descendants);
+ this.conf = conf;
+ }
+
+ @Override
+ protected String property(String key) {
+ return conf.get(key);
+ }
+
+ @Override
+ protected Provider provider(String scheme) {
+ if ("file".equalsIgnoreCase(scheme)) {
+ return Provider.LOCAL;
+ }
+ try {
+ String implementation = FileSystem.getFileSystemClass(scheme,
conf).getName();
+ if (Arrays.asList("io.juicefs.JuiceFileSystem",
"org.apache.hadoop.fs.ozone.OzoneFileSystem",
+ "org.apache.hadoop.fs.ozone.RootedOzoneFileSystem",
"org.apache.hadoop.fs.cosn.CosNFileSystem",
+
"com.qcloud.chdfs.fs.CHDFSHadoopFileSystemAdapter").contains(implementation)) {
+ return Provider.REMOTE;
+ }
+ return implementationProvider(implementation);
+ } catch (IOException | RuntimeException e) {
+ return Provider.UNKNOWN;
+ }
+ }
+ }
+
+ public static class Route {
+ protected enum Provider {
+ REMOTE,
+ LOCAL,
+ MOUNT_TABLE,
+ DYNAMIC,
+ UNKNOWN
+ }
+
+ protected static final Set<String> REMOTE_SCHEMES = new
HashSet<>(Arrays.asList(
+ "hdfs", "s3a", "s3n", "oss", "ofs", "jfs", "abfs", "abfss", "wasb",
"wasbs", "gs", "cosn"));
+ protected static final Set<String> REMOTE_IMPLEMENTATIONS = new
HashSet<>(Arrays.asList(
+ "org.apache.hadoop.hdfs.DistributedFileSystem",
+ "org.apache.hadoop.fs.s3a.S3AFileSystem",
+ "org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem",
+ "org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem",
+ "org.apache.hadoop.fs.azure.NativeAzureFileSystem",
Review Comment:
Fixed in fb92e7d8c337c86e6e90fe135bbd4a5d995115b0. The shared exact-provider
set includes Hadoop's standard NativeAzureFileSystem$Secure. Controls cover
actual default/inherited direct and delegated WASBS routes, plus local override
rejection. The same fix is synchronized across all three PRs.
Validation: Final standard JDK17 FE: 107 selected tests passed with fresh
XML, zero failures/errors/skips and frozen source. Final standard Clang21/ASAN
BE: 28 selected tests passed. Each release also passed 16 Lance and 19 Paimon
build-control tests; the new SDK unit/C API tests, matching Rust std/ABI checks
and private standard build passed. Independent Standards/Spec rechecks found no
remaining confirmed issue in this follow-up.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -22,24 +22,218 @@
#include <arrow/record_batch.h>
#include <arrow/type.h>
#include <lance/lance.h>
+#include <pwd.h>
+#include <rapidjson/document.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"
+#ifndef LANCE_STORAGE_BASES_JSON_VERSION
+#error "Rebuild lance-c with TVF storage-base metadata support before building
Doris"
+#endif
+
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();
+}
+
+bool is_local_lance_tvf(const TFileRangeDesc& range, const
TFileScanRangeParams& scan_params) {
+ const bool local = range.__isset.file_type
+ ? range.file_type == TFileType::FILE_LOCAL
+ : scan_params.__isset.file_type &&
+ scan_params.file_type ==
TFileType::FILE_LOCAL;
+ const auto& params = range.table_format_params.lance_params;
+ return local && (params.version == 0 || (params.__isset.is_tvf &&
params.is_tvf));
+}
+
+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;
+ // Explicit TVF identity applies regardless of snapshot selection. Older
LOCAL TVFs
+ // used version zero; configured fixed-version catalogs do not send the
TVF marker.
+ if (is_local_lance_tvf(range, scan_params)) {
+ // 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(
Review Comment:
Fixed in fb92e7d8c337c86e6e90fe135bbd4a5d995115b0. Unsupported LOCAL URI
aliases now use a small pre-open SDK metadata API that reuses the pinned SDK's
own URL parser/default registry path extraction without constructing a store or
reading a manifest. BE canonicalizes and checks the extracted path and
reachable tree, then reopens the canonical URI with the original local provider
and percent-encoded path. It does not rewrite the alias to file: before
parsing. Established file: URI parsing stays unchanged. Actual schema and
row-value controls cover mandatory file-object-store, encoded
space/percent/plus, authority/query/fragment, in-root symlink, and out-of-root
rejection; the dataset key preserves provider identity. Extra file+uring scans
are explicitly opt-in and not claimed here because this host denies io_uring
setup. Matching header/archive macros and defined exports are enforced.
Validation: Final standard JDK17 FE: 107 selected tests passed with fresh
XML, zero failures/errors/skips and frozen source. Final standard Clang21/ASAN
BE: 28 selected tests passed. Each release also passed 16 Lance and 19 Paimon
build-control tests; the new SDK unit/C API tests, matching Rust std/ABI checks
and private standard build passed. Independent Standards/Spec rechecks found no
remaining confirmed issue in this follow-up.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java:
##########
@@ -135,6 +143,38 @@ private void getFileListFromBackend() throws
AnalysisException {
}
}
+ /** Apply LOCAL admission to sinks that select the built-in local JNI
writer as well. */
+ public static void checkSinkEnabled(String tvfName, Map<String, String>
properties) {
+ String writerClass = properties.getOrDefault("writer_class", "");
+ // JNI consumes a C string, and ScannerLoader accepts slash-separated
class names.
+ int nul = writerClass.indexOf('\0');
+ if (nul >= 0) {
+ writerClass = writerClass.substring(0, nul);
+ }
+ if ("local".equalsIgnoreCase(tvfName)
Review Comment:
Fixed in fb92e7d8c337c86e6e90fe135bbd4a5d995115b0. The retained nested
ViewFS variant is confirmed and fixed: guarded FrozenConfiguration forces
fs.viewfs.enable.inner.cache=false and includes that effective value in its
profile. Nested children therefore use the existing FileSystem cache, whose
guarded identity preserves exact authority and configuration. The folding
InnerCache can no longer reuse A for a; ordinary unmarked requests keep their
configured cache behavior. Native write admission still checks the actual path
before IO and enabled-write opt-in is unchanged.
Validation: Final standard JDK17 FE: 107 selected tests passed with fresh
XML, zero failures/errors/skips and frozen source. Final standard Clang21/ASAN
BE: 28 selected tests passed. Each release also passed 16 Lance and 19 Paimon
build-control tests; the new SDK unit/C API tests, matching Rust std/ABI checks
and private standard build passed. Independent Standards/Spec rechecks found no
remaining confirmed issue in this follow-up.
--
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]
