diff --git a/be/src/format_v2/table/lance_reader.cpp b/be/src/format_v2/table/lance_reader.cpp index e0e2884efe7a4a..c1db3b5f98b322 100644 --- a/be/src/format_v2/table/lance_reader.cpp +++ b/be/src/format_v2/table/lance_reader.cpp @@ -35,11 +35,50 @@ #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 { +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) / requested_uri).lexically_normal().string(); + } + return io::global_local_filesystem()->canonicalize_local_file( + config::user_files_secure_path, local_input, uri); + } + *uri = requested_uri; + return Status::OK(); +} +} // namespace + LanceTableReader::~LanceTableReader() { static_cast(close()); } @@ -52,6 +91,8 @@ Status LanceTableReader::fetch_schema(const TFileRangeDesc& range, return Status::InvalidArgument("Lance schema output must not be null"); } const auto& params = range.table_format_params.lance_params; + std::string dataset_uri; + RETURN_IF_ERROR(resolve_dataset_uri(range, scan_params, &dataset_uri)); std::vector storage_options; RETURN_IF_ERROR(build_lance_storage_options(&scan_params, &storage_options)); std::vector storage_option_ptrs; @@ -61,10 +102,9 @@ Status LanceTableReader::fetch_schema(const TFileRangeDesc& range, } storage_option_ptrs.emplace_back(nullptr); - std::unique_ptr dataset( - lance_dataset_open(params.dataset_uri.c_str(), - storage_options.empty() ? nullptr : storage_option_ptrs.data(), - static_cast(params.version))); + std::unique_ptr dataset(lance_dataset_open( + dataset_uri.c_str(), storage_options.empty() ? nullptr : storage_option_ptrs.data(), + static_cast(params.version))); if (dataset == nullptr) { return lance_error("open Lance dataset for schema"); } @@ -1329,7 +1369,7 @@ Status LanceTableReader::_fill_block_from_lance_batch(LanceBatch* batch, Block* Status LanceTableReader::_dataset_key(const TFileRangeDesc& range, DatasetKey* key) const { const auto& params = range.table_format_params.lance_params; - key->uri = params.dataset_uri; + RETURN_IF_ERROR(resolve_dataset_uri(range, *_scan_params, &key->uri)); key->version = params.version; return build_lance_storage_options(_scan_params, &key->storage_options); } diff --git a/be/src/io/fs/local_file_system.cpp b/be/src/io/fs/local_file_system.cpp index 4f9b6baf2a75cf..d51ecb6decf4d1 100644 --- a/be/src/io/fs/local_file_system.cpp +++ b/be/src/io/fs/local_file_system.cpp @@ -455,14 +455,128 @@ const std::shared_ptr& global_local_filesystem() { Status LocalFileSystem::canonicalize_local_file(const std::string& dir, const std::string& file_path, std::string* full_path) { - const std::string absolute_path = dir + "/" + file_path; + // An empty root would turn a relative request into an unrestricted absolute path. + if (dir.empty()) { + return Status::InvalidArgument("user_files_secure_path must not be empty"); + } + // Reject embedded NULs before passing paths to filesystem APIs that use C strings. + if (file_path.empty() || file_path.find('\0') != std::string::npos) { + return Status::InvalidArgument("local file path must not be empty or contain NUL"); + } + // Match URL preprocessing only for actual file URIs; bare paths keep literal whitespace. + const auto is_c0_or_space = [](unsigned char c) { return c <= 0x20; }; + size_t begin = 0; + size_t end = file_path.size(); + while (begin < end && is_c0_or_space(file_path[begin])) { + ++begin; + } + while (end > begin && is_c0_or_space(file_path[end - 1])) { + --end; + } + std::string uri_input; + for (size_t i = begin; i < end; ++i) { + if (file_path[i] != '\t' && file_path[i] != '\r' && file_path[i] != '\n') { + uri_input += file_path[i]; + } + } + const bool is_file_uri = uri_input.size() >= 5 && + (uri_input[0] == 'f' || uri_input[0] == 'F') && + (uri_input[1] == 'i' || uri_input[1] == 'I') && + (uri_input[2] == 'l' || uri_input[2] == 'L') && + (uri_input[3] == 'e' || uri_input[3] == 'E') && uri_input[4] == ':'; + std::string local_input = is_file_uri ? std::move(uri_input) : file_path; + if (is_file_uri) { + local_input.replace(0, 4, "file"); + // File URLs treat raw backslashes as separators, including before the authority. + // Do this before conversion; percent-encoded backslashes must remain literal. + for (size_t i = 5; i < local_input.size(); ++i) { + if (local_input[i] == '\\') { + local_input[i] = '/'; + } + } + } + Path local_path; + // Convert file: URIs to filesystem paths and reject remote schemes before any reader opens them. + RETURN_IF_ERROR(convert_to_abs_path(local_input, local_path)); + if (is_file_uri) { + // file:a%20b names "a b", but a bare path containing "%20" must stay literal. + // Decode once, preserve '+' and invalid escapes, matching Lance's file provider. + auto encoded = local_path.string(); + // Query and fragment are URI components, not part of the filesystem path. + encoded = encoded.substr(0, encoded.find_first_of("?#")); + // A file URL without a leading slash still names a filesystem-rooted path. + if (encoded.empty() || encoded[0] != '/') { + encoded.insert(0, "/"); + } + // URL dot segments are removed before symlinks are followed and before %2F + // becomes a path separator. Keep empty segments: "a//../b" denotes "a/b". + std::vector segments; + for (size_t start = 1;;) { + const size_t end = encoded.find('/', start); + const auto segment = + encoded.substr(start, end == std::string::npos ? end : end - start); + const bool dot = segment == "." || segment == "%2e" || segment == "%2E"; + const bool parent = segment == ".." || segment == ".%2e" || segment == ".%2E" || + segment == "%2e." || segment == "%2E." || segment == "%2e%2e" || + segment == "%2e%2E" || segment == "%2E%2e" || segment == "%2E%2E"; + if (parent) { + if (!segments.empty()) { + segments.pop_back(); + } + } else if (!dot) { + segments.push_back(segment); + } + if (end == std::string::npos) { + if (dot || parent) { + segments.emplace_back(""); + } + break; + } + start = end + 1; + } + encoded = "/"; + for (size_t i = 0; i < segments.size(); ++i) { + if (i > 0) { + encoded += '/'; + } + encoded += segments[i]; + } + const auto hex_digit = [](char c) -> int { + if (c >= '0' && c <= '9') return c - '0'; + if (c >= 'a' && c <= 'f') return c - 'a' + 10; + if (c >= 'A' && c <= 'F') return c - 'A' + 10; + return -1; + }; + std::string decoded; + decoded.reserve(encoded.size()); + for (size_t i = 0; i < encoded.size(); ++i) { + if (encoded[i] == '%' && i + 2 < encoded.size()) { + const int high = hex_digit(encoded[i + 1]); + const int low = hex_digit(encoded[i + 2]); + if (high >= 0 && low >= 0) { + decoded += static_cast((high << 4) | low); + i += 2; + continue; + } + } + decoded += encoded[i]; + } + // %00 must not truncate the path when canonicalization reaches C filesystem APIs. + if (decoded.find('\0') != std::string::npos) { + return Status::InvalidArgument("local file path must not contain NUL"); + } + local_path = std::move(decoded); + } + std::string secure_path; + RETURN_IF_ERROR(canonicalize(dir, &secure_path)); std::string canonical_path; - RETURN_IF_ERROR(canonicalize(absolute_path, &canonical_path)); - if (!contain_path(dir, canonical_path)) { - return Status::InvalidArgument("file path is not allowed: {}", canonical_path); + const auto status = canonicalize(Path(secure_path) / local_path, &canonical_path); + // A glob result can reveal external names even when resolution fails on a dangling symlink. + // Reject both failed resolution and out-of-root targets without exposing expanded names. + if (!status.ok() || !contain_path(secure_path, canonical_path)) { + return Status::InvalidArgument("file path is not allowed"); } - - *full_path = canonical_path; + *full_path = std::move(canonical_path); return Status::OK(); } @@ -470,13 +584,29 @@ Status LocalFileSystem::safe_glob(const std::string& path, std::vector if (path.find("..") != std::string::npos) { return Status::InvalidArgument("can not contain '..' in path"); } - std::string full_path = config::user_files_secure_path + "/" + path; + // An empty root would turn the prefix below into "/" and allow any local file. + if (config::user_files_secure_path.empty()) { + return Status::InvalidArgument("user_files_secure_path must not be empty"); + } + std::string secure_path; + RETURN_IF_ERROR(canonicalize(config::user_files_secure_path, &secure_path)); + // A symlink root may canonicalize to a directory such as "user_files[prod]". + // The root is literal; only the user-supplied suffix may contain glob operators. + std::string full_path; + for (char c : secure_path) { + if (c == '\\' || c == '*' || c == '?' || c == '[' || c == ']') { + full_path += '\\'; + } + full_path += c; + } + full_path += "/" + path; std::vector files; RETURN_IF_ERROR(_glob(full_path, &files)); for (auto& file : files) { FileInfo fi; fi.is_file = true; - RETURN_IF_ERROR(canonicalize_local_file("", file, &(fi.file_name))); + // Glob classes and symlinks can escape without a literal ".." in the pattern. + RETURN_IF_ERROR(canonicalize_local_file(secure_path, file, &(fi.file_name))); RETURN_IF_ERROR(file_size_impl(fi.file_name, &(fi.file_size))); res->push_back(std::move(fi)); } diff --git a/be/src/io/fs/local_file_system.h b/be/src/io/fs/local_file_system.h index b1546bc473426f..1f7ffc8da51964 100644 --- a/be/src/io/fs/local_file_system.h +++ b/be/src/io/fs/local_file_system.h @@ -70,6 +70,8 @@ class LocalFileSystem final : public FileSystem { static std::filesystem::perms PERMS_OWNER_RW; + // Resolve a local path relative to a nonempty secure root, then check canonical containment. + // Absolute paths and file: URIs must resolve inside the same root; remote schemes are rejected. Status canonicalize_local_file(const std::string& dir, const std::string& file_path, std::string* full_path); diff --git a/be/test/format_v2/table/lance_reader_test.cpp b/be/test/format_v2/table/lance_reader_test.cpp index 6531fc21de72e8..6aa3b32a8bb3ba 100644 --- a/be/test/format_v2/table/lance_reader_test.cpp +++ b/be/test/format_v2/table/lance_reader_test.cpp @@ -49,6 +49,7 @@ #include #include +#include "common/config.h" #include "common/consts.h" #include "core/assert_cast.h" #include "core/block/block.h" @@ -4313,3 +4314,262 @@ TEST(LanceTableReaderTypeTest, ReadsTemporalTypesFromAllTypesFixture) { } } // namespace doris::format::lance + +namespace doris::format::lance { +TEST(LanceTableReaderLocalPathTest, SchemaAndScanUseTheSameSecurePath) { + const auto original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + const auto root = std::filesystem::canonical("./be/test/format_v2/table/lance/data"); + config::user_files_secure_path = root.string(); + TFileScanRangeParams scan_params; + LanceTableReader reader; + reader._scan_params = &scan_params; + auto range = make_latest_lance_range("all_types.lance"); + range.__set_file_type(TFileType::FILE_LOCAL); + std::vector names; + std::vector types; + ASSERT_TRUE(reader.fetch_schema(range, scan_params, &names, &types).ok()); + EXPECT_FALSE(names.empty()); + LanceTableReader::DatasetKey key; + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ((root / "all_types.lance").string(), key.uri); + + // Legacy scan-level tags must enforce the same restriction as per-range tags. + range.__isset.file_type = false; + scan_params.__set_file_type(TFileType::FILE_LOCAL); + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ((root / "all_types.lance").string(), key.uri); + + for (const auto& uri : std::vector { + "https://invalid/dataset", "http://invalid/dataset", "s3://bucket/dataset", + "../data-other/dataset", "", std::string("a\0b", 3) + "/../all_types.lance"}) { + range.table_format_params.lance_params.__set_dataset_uri(uri); + const auto schema_status = reader.fetch_schema(range, scan_params, &names, &types); + const auto scan_status = reader._dataset_key(range, &key); + EXPECT_FALSE(schema_status.ok()) << uri; + EXPECT_EQ(schema_status.to_string(), scan_status.to_string()); + } +} + +TEST(LanceTableReaderLocalPathTest, ScansRelativeDatasetInsideSecureRoot) { + const auto original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + config::user_files_secure_path = + std::filesystem::canonical("./be/test/format_v2/table/lance/data").string(); + TQueryGlobals query_globals; + RuntimeState state(query_globals); + RuntimeProfile profile("lance_secure_local_scan"); + TFileScanRangeParams scan_params; + const Columns columns {projected_column("row_id", TYPE_BIGINT, false)}; + LanceTableReader reader; + ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params).ok()); + auto range = make_latest_lance_range("all_types.lance"); + range.__set_file_type(TFileType::FILE_LOCAL); + ASSERT_TRUE(prepare_range(&reader, range).ok()); + + // Exercise data reads through the validated relative LOCAL range. + Block block; + add_output_columns(&block, columns); + std::vector row_ids; + bool eos = false; + while (!eos) { + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + if (eos) { + continue; + } + const auto& values = + assert_cast(*block.get_by_position(0).column).get_data(); + for (size_t row = 0; row < block.rows(); ++row) { + row_ids.push_back(values[row]); + } + } + std::sort(row_ids.begin(), row_ids.end()); + EXPECT_EQ((std::vector {1, 2, 3, 4}), row_ids); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(LanceTableReaderLocalPathTest, FileUrisPreserveEscapedDatasetIdentity) { + const auto original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + const auto root = std::filesystem::temp_directory_path() / + ("lance_escaped_uri_" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + DEFER({ + std::error_code error; + std::filesystem::remove_all(root, error); + }); + const auto fixture = + std::filesystem::canonical("./be/test/format_v2/table/lance/data/all_types.lance"); + for (const auto* name : {"a b.lance", "a%20b.lance", "plus+percent%.lance"}) { + std::filesystem::create_directories(root / name); + std::filesystem::copy(fixture, root / name, std::filesystem::copy_options::recursive); + } + config::user_files_secure_path = root.string(); + for (const auto& input : + {std::make_pair("file://" + (root / "a%20b.lance").string(), root / "a b.lance"), + std::make_pair(std::string("a%20b.lance"), root / "a%20b.lance"), + std::make_pair("file://" + (root / "a%20b.lance").string() + "?q=%00#tag", + root / "a b.lance"), + std::make_pair("file://" + (root / "a%2520b.lance").string(), root / "a%20b.lance"), + std::make_pair("file://" + (root / "plus+percent%25.lance").string(), + root / "plus+percent%.lance")}) { + TQueryGlobals query_globals; + RuntimeState state(query_globals); + RuntimeProfile profile("lance_escaped_file_uri"); + TFileScanRangeParams scan_params; + LanceTableReader reader; + const Columns columns = {projected_column("row_id", TYPE_BIGINT, false)}; + ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params).ok()); + auto range = make_latest_lance_range(input.first); + range.__set_file_type(TFileType::FILE_LOCAL); + std::vector names; + std::vector types; + ASSERT_TRUE(reader.fetch_schema(range, scan_params, &names, &types).ok()) << input.first; + EXPECT_FALSE(names.empty()); + LanceTableReader::DatasetKey key; + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ(input.second.string(), key.uri); + ASSERT_TRUE(prepare_range(&reader, range).ok()); + Block block; + add_output_columns(&block, columns); + std::vector row_ids; + bool eos = false; + while (!eos) { + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + if (!eos) { + const auto& values = + assert_cast(*block.get_by_position(0).column) + .get_data(); + for (size_t row = 0; row < block.rows(); ++row) { + row_ids.push_back(values[row]); + } + } + } + std::sort(row_ids.begin(), row_ids.end()); + EXPECT_EQ((std::vector {1, 2, 3, 4}), row_ids); + ASSERT_TRUE(reader.close().ok()); + } +} + +TEST(LanceTableReaderLocalPathTest, FileUrisNormalizeBeforeFilesystemResolution) { + const auto original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + const auto root = std::filesystem::temp_directory_path() / + ("lance_uri_normalization_" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + DEFER({ + std::error_code error; + std::filesystem::remove_all(root, error); + }); + std::filesystem::create_directories(root / "nested/child"); + std::filesystem::create_directory_symlink(root / "nested/child", root / "jump"); + std::filesystem::create_directory_symlink(root, root / "secure-link"); + const auto backslash_uri = std::string(R"(fIlE:\\localhost)") + root.string() + "/data.lance"; + struct DatasetFixture { + const char* column_name; + int64_t row_id; + std::filesystem::path path; + }; + const DatasetFixture datasets[] = { + {"root_only", 7, root / "data.lance"}, + {"nested_only", 42, root / "nested/data.lance"}, + {"spaced_only", 99, root / "data.lance "}, + {"prefixed_only", 123, root / (root / "data.lance").relative_path()}}; + for (const auto& dataset : datasets) { + std::filesystem::create_directories(dataset.path.parent_path()); + arrow::Int64Builder values_builder; + ASSERT_TRUE(values_builder.Append(dataset.row_id).ok()); + std::shared_ptr values; + ASSERT_TRUE(values_builder.Finish(&values).ok()); + const auto schema = + arrow::schema({arrow::field("row_id", arrow::int64(), false), + arrow::field(dataset.column_name, arrow::int64(), false)}); + ASSERT_TRUE(write_lance_record_batch(dataset.path, + arrow::RecordBatch::Make(schema, 1, {values, values})) + .ok()); + } + config::user_files_secure_path = (root / "secure-link").string(); + for (const auto& input : + {std::make_pair("file://" + root.string() + "/jump/../data.lance", &datasets[0]), + std::make_pair("file://" + root.string() + "/missing/../data.lance", &datasets[0]), + std::make_pair("file://" + root.string() + "/jump/%2e%2E/data.lance", &datasets[0]), + std::make_pair("FILE://" + root.string() + "/data.lance", &datasets[0]), + std::make_pair(backslash_uri, &datasets[0]), + std::make_pair("fIlE://" + root.string() + "/jump/.%2e/data.lance", &datasets[0]), + std::make_pair("file://" + root.string() + "/jump%2f../data.lance", &datasets[1]), + std::make_pair(std::string("jump/../data.lance"), &datasets[0]), + std::make_pair((root / "jump/../data.lance").string(), &datasets[0]), + std::make_pair((root / "missing/../data.lance").string(), &datasets[0]), + std::make_pair("file:" + (root / "data.lance").relative_path().string(), &datasets[0]), + std::make_pair("file://" + root.string() + "/data.lance ", &datasets[0]), + std::make_pair(" \tFi\nLe://" + root.string() + "/data.la\tnce\r\n ", &datasets[0]), + std::make_pair("file://" + root.string() + "/data.lance%20", &datasets[2]), + std::make_pair("file://" + root.string() + "/data.lance%20 ", &datasets[2]), + std::make_pair(std::string("data.lance "), &datasets[2]), + std::make_pair((root / "data.lance ").string(), &datasets[2])}) { + SCOPED_TRACE("[" + input.first + "]"); + const auto& expected = *input.second; + TQueryGlobals globals; + RuntimeState state(globals); + RuntimeProfile profile("lance_uri_normalization"); + TFileScanRangeParams scan_params; + LanceTableReader reader; + const Columns columns {projected_column("row_id", TYPE_BIGINT, false)}; + ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params).ok()); + auto range = make_latest_lance_range(input.first); + range.__set_file_type(TFileType::FILE_LOCAL); + std::vector names; + std::vector types; + ASSERT_TRUE(reader.fetch_schema(range, scan_params, &names, &types).ok()); + EXPECT_EQ((std::vector {"row_id", expected.column_name}), names); + LanceTableReader::DatasetKey key; + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ(expected.path.string(), key.uri); + ASSERT_TRUE(prepare_range(&reader, range).ok()); + Block block; + add_output_columns(&block, columns); + std::vector row_ids; + bool eos = false; + while (!eos) { + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + if (!eos) { + const auto& values = + assert_cast(*block.get_by_position(0).column) + .get_data(); + row_ids.insert(row_ids.end(), values.begin(), values.end()); + } + } + EXPECT_EQ((std::vector {expected.row_id}), row_ids); + ASSERT_TRUE(reader.close().ok()); + } +} + +TEST(LanceTableReaderLocalPathTest, FixedVersionLocalCatalogKeepsItsUri) { + TFileScanRangeParams scan_params; + LanceTableReader reader; + reader._scan_params = &scan_params; + // Catalog warehouses use their own location, not the LOCAL TVF secure directory. + auto range = make_lance_range("file:///catalog-owned/dataset.lance", 7, {0}); + range.__set_file_type(TFileType::FILE_LOCAL); + LanceTableReader::DatasetKey key; + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ("file:///catalog-owned/dataset.lance", key.uri); + EXPECT_EQ(7, key.version); +} + +TEST(LanceTableReaderLocalPathTest, RemoteRangeKeepsItsUriAndOptions) { + TFileScanRangeParams scan_params; + scan_params.__set_file_type(TFileType::FILE_LOCAL); + TLanceScanParams lance_params; + lance_params.__set_lance_storage_options({{"aws_region", "us-east-1"}}); + scan_params.__set_lance_scan_params(lance_params); + LanceTableReader reader; + reader._scan_params = &scan_params; + auto range = make_latest_lance_range("s3://bucket/dataset.lance"); + range.__set_file_type(TFileType::FILE_S3); + LanceTableReader::DatasetKey key; + ASSERT_TRUE(reader._dataset_key(range, &key).ok()); + EXPECT_EQ("s3://bucket/dataset.lance", key.uri); + EXPECT_EQ((std::vector {"aws_region", "us-east-1"}), key.storage_options); +} +} // namespace doris::format::lance diff --git a/be/test/io/fs/local_file_system_test.cpp b/be/test/io/fs/local_file_system_test.cpp index 4321a2ae56290f..09660e2d84741c 100644 --- a/be/test/io/fs/local_file_system_test.cpp +++ b/be/test/io/fs/local_file_system_test.cpp @@ -30,12 +30,14 @@ #include #include +#include "common/config.h" #include "common/status.h" #include "cpp/sync_point.h" #include "gtest/gtest_pred_impl.h" #include "io/fs/file_reader.h" #include "io/fs/file_writer.h" #include "io/fs/local_file_writer.h" +#include "util/defer_op.h" #include "util/slice.h" namespace doris { @@ -547,6 +549,10 @@ TEST_F(LocalFileSystemTest, AbnormalWriteRead) { } TEST_F(LocalFileSystemTest, TestGlob) { + const std::string original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + // Bind this fixture to its own files, independent of the runner's build-profile directory. + config::user_files_secure_path = std::filesystem::absolute("./be/ut_build_ASAN/test").string(); std::string path = "./be/ut_build_ASAN/test/file_path/"; EXPECT_TRUE(io::global_local_filesystem()->delete_directory(path).ok()); EXPECT_TRUE(io::global_local_filesystem() @@ -578,6 +584,205 @@ TEST_F(LocalFileSystemTest, TestGlob) { EXPECT_TRUE(io::global_local_filesystem()->delete_directory(path).ok()); } +TEST_F(LocalFileSystemTest, TestSafeGlobPathContainment) { + auto fs = io::global_local_filesystem(); + const std::string original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + const auto root = std::filesystem::canonical(test_dir); + const auto secure = root / "secure"; + const auto outside = root / "secure-other"; + std::filesystem::create_directories(secure / "nested"); + std::filesystem::create_directory(outside); + ASSERT_TRUE(save_string_file((secure / "nested/inside.csv").string(), "inside").ok()); + ASSERT_TRUE(save_string_file((outside / "outside.csv").string(), "outside").ok()); + std::filesystem::create_directory_symlink(outside, secure / "escape"); + std::filesystem::create_symlink(outside / "outside.csv", secure / "outside.csv"); + std::filesystem::create_symlink(outside / "missing.csv", outside / "broken.csv"); + std::filesystem::create_directory_symlink(secure / "nested", secure / "inside"); + config::user_files_secure_path = secure.string(); + + for (const auto* pattern : {".[.]/secure-other/outside.csv", ".?/secure-other/outside.csv", + ".[.-.]/secure-other/outside.csv", "escape/*.csv", "outside.csv", + ".[.]/secure-other/*.csv", "escape/broken*.csv"}) { + std::vector files; + const auto status = fs->safe_glob(pattern, &files); + EXPECT_FALSE(status.ok()) << pattern; + EXPECT_NE(status.to_string().find("file path is not allowed"), std::string::npos) << status; + // The pattern need not name the matched file, so errors must not reveal that name. + const auto message = status.to_string(); + EXPECT_EQ(std::string::npos, message.find(root.string())) << message; + for (const auto* name : {"outside.csv", "broken.csv", "missing.csv"}) { + EXPECT_EQ(std::string::npos, message.find(name)) << message; + } + EXPECT_TRUE(files.empty()); + } + + for (const auto* pattern : {"nested/*.csv", "inside/*.csv"}) { + std::vector files; + ASSERT_TRUE(fs->safe_glob(pattern, &files).ok()) << pattern; + ASSERT_EQ(1, files.size()); + EXPECT_EQ((secure / "nested/inside.csv").string(), files[0].file_name); + EXPECT_EQ(6, files[0].file_size); + } +} + +TEST_F(LocalFileSystemTest, TestSafeGlobSecureRoot) { + auto fs = io::global_local_filesystem(); + const std::string original_secure_path = config::user_files_secure_path; + DEFER(config::user_files_secure_path = original_secure_path); + const auto root = std::filesystem::canonical(test_dir); + std::filesystem::create_directory(root / "secure"); + std::filesystem::create_directory_symlink(root / "secure", root / "secure-link"); + ASSERT_TRUE(save_string_file((root / "secure/inside.csv").string(), "inside").ok()); + + for (const auto& secure : {std::string(test_dir) + "/secure", (root / "secure/").string(), + (root / "secure-link").string()}) { + config::user_files_secure_path = secure; + std::vector files; + ASSERT_TRUE(fs->safe_glob("*.csv", &files).ok()) << secure; + ASSERT_EQ(1, files.size()); + EXPECT_EQ((root / "secure/inside.csv").string(), files[0].file_name); + } + const auto special_root = root / "secure[prod]*?\\root"; + std::filesystem::create_directory(special_root); + std::filesystem::create_directory_symlink(special_root, root / "special-root-link"); + ASSERT_TRUE(save_string_file((special_root / "inside.csv").string(), "inside").ok()); + for (const auto& secure : {special_root, root / "special-root-link"}) { + config::user_files_secure_path = secure.string(); + std::vector matches; + ASSERT_TRUE(fs->safe_glob("*.csv", &matches).ok()) << secure; + ASSERT_EQ(1, matches.size()); + EXPECT_EQ((special_root / "inside.csv").string(), matches[0].file_name); + } + config::user_files_secure_path = ""; + std::vector files; + EXPECT_FALSE(fs->safe_glob("*.csv", &files).ok()); + config::user_files_secure_path = (root / "missing").string(); + EXPECT_FALSE(fs->safe_glob("*.csv", &files).ok()); +} + +TEST_F(LocalFileSystemTest, TestCanonicalizeSecureLocalPath) { + auto fs = io::global_local_filesystem(); + const auto root = std::filesystem::canonical(test_dir); + const auto secure = root / "secure"; + std::filesystem::create_directories(secure / "nested"); + std::filesystem::create_directory(root / "secure-other"); + std::filesystem::create_directory_symlink(root / "secure-other", secure / "escape"); + std::filesystem::create_directory_symlink(secure / "nested", secure / "inside"); + for (const auto& path : + {std::string("nested"), std::string("nested/../inside"), (secure / "nested").string(), + "file://" + (secure / "nested").string()}) { + std::string resolved; + ASSERT_TRUE(fs->canonicalize_local_file(secure.string(), path, &resolved).ok()) << path; + EXPECT_EQ((secure / "nested").string(), resolved); + } + for (const auto& path : + {std::string("../secure-other"), std::string("escape"), (root / "secure-other").string(), + std::string("https://invalid/dataset"), std::string("s3://bucket/dataset"), + std::string("http://invalid/dataset"), std::string(""), + std::string("nested\0ignored", 14)}) { + std::string resolved; + const auto status = fs->canonicalize_local_file(secure.string(), path, &resolved); + EXPECT_FALSE(status.ok()) << path; + EXPECT_EQ(std::string::npos, status.to_string().find(root.string())) << status; + } + // URI escapes and bare '%' paths may name two different, simultaneously existing datasets. + for (const auto* name : {"a b.lance", "a%20b.lance", "plus+percent%.lance", "bad%2G.lance"}) { + std::filesystem::create_directory(secure / name); + } + for (const auto& input : + {std::make_pair("file://" + (secure / "a%20b.lance").string(), secure / "a b.lance"), + std::make_pair(std::string("a%20b.lance"), secure / "a%20b.lance"), + std::make_pair("file://" + (secure / "a%20b.lance").string() + "#tag", + secure / "a b.lance"), + std::make_pair("file://" + (secure / "a%20b.lance").string() + "?q=%00#tag", + secure / "a b.lance"), + std::make_pair("file://" + (secure / "a%2520b.lance").string(), secure / "a%20b.lance"), + std::make_pair("file://" + (secure / "plus+percent%25.lance").string(), + secure / "plus+percent%.lance"), + std::make_pair("file://" + (secure / "bad%2G.lance").string(), + secure / "bad%2G.lance")}) { + std::string actual; + ASSERT_TRUE(fs->canonicalize_local_file(secure.string(), input.first, &actual).ok()) + << input.first; + EXPECT_EQ(input.second.string(), actual); + } + for (const auto& suffix : + {"/%2e%2e/secure-other", "/nested%2f..%2f..%2fsecure-other", "/a%00b.lance"}) { + std::string actual; + EXPECT_FALSE(fs->canonicalize_local_file(secure.string(), + "file://" + secure.string() + suffix, &actual) + .ok()); + } + std::string resolved; + EXPECT_FALSE(fs->canonicalize_local_file("", secure.string(), &resolved).ok()); +} + +TEST_F(LocalFileSystemTest, TestCanonicalizeFileUriIdentity) { + auto fs = io::global_local_filesystem(); + const auto root = std::filesystem::canonical(test_dir); + const auto secure = root / "secure"; + std::filesystem::create_directories(secure / "nested/child/data.lance"); + std::filesystem::create_directory(secure / "nested/data.lance"); + std::filesystem::create_directory(secure / "data.lance"); + std::filesystem::create_directory(secure / "a b.lance"); + std::filesystem::create_directory(secure / "data.lance "); + std::filesystem::create_directories(secure / (secure / "data.lance").relative_path()); + std::filesystem::create_directory(root / "outside"); + std::filesystem::create_directory_symlink(secure / "nested/child", secure / "jump"); + std::filesystem::create_directory(secure / "literal\\name.lance"); + const auto backslash_uri = std::string(R"(fIlE:\\localhost)") + secure.string() + "/data.lance"; + for (const auto& input : + {std::make_pair("file://" + secure.string() + "/jump/../data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/missing/../data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/jump/%2e%2E/data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/jump/.%2e/data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/jump/%2E./data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/missing/.././data.lance", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/jump//../data.lance", + secure / "nested/child/data.lance"), + std::make_pair("file://" + secure.string() + "/jump%2f../data.lance", + secure / "nested/data.lance"), + std::make_pair("file://" + secure.string() + "/jump\\..\\data.lance", + secure / "data.lance"), + std::make_pair("FILE://" + secure.string() + "/data.lance", secure / "data.lance"), + std::make_pair(backslash_uri, secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/literal%5cname.lance", + secure / "literal\\name.lance"), + std::make_pair(std::string("literal\\name.lance"), secure / "literal\\name.lance"), + std::make_pair("fIlE://" + secure.string() + "/a%20b.lance?q=%00#tag", + secure / "a b.lance"), + std::make_pair("file:" + (secure / "data.lance").relative_path().string(), + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/data.lance ", secure / "data.lance"), + std::make_pair(" \tFi\nLe://" + secure.string() + "/data.la\tnce\r\n ", + secure / "data.lance"), + std::make_pair("file://" + secure.string() + "/data.lance%20", secure / "data.lance "), + std::make_pair("file://" + secure.string() + "/data.lance%20 ", secure / "data.lance "), + std::make_pair(std::string("data.lance "), secure / "data.lance "), + std::make_pair(std::string("jump/../data.lance"), secure / "nested/data.lance")}) { + std::string actual; + const auto status = fs->canonicalize_local_file(secure.string(), input.first, &actual); + ASSERT_TRUE(status.ok()) << input.first << " " << status; + EXPECT_EQ(input.second.string(), actual) << input.first; + } + for (const auto& input : + {"file://" + secure.string() + "/jump/../../outside", + "FILE://" + secure.string() + "/%2e%2e/outside", + "fIlE://" + secure.string() + "/a%00b.lance", std::string("HTTP://invalid/dataset")}) { + std::string actual; + const auto status = fs->canonicalize_local_file(secure.string(), input, &actual); + EXPECT_FALSE(status.ok()) << input; + EXPECT_EQ(std::string::npos, status.to_string().find(root.string())); + } +} + TEST_F(LocalFileSystemTest, TestConvertToAbsPath) { io::Path abs_path; Status st; diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index e1fc70a673362b..366e9af2ac4afd 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -3101,6 +3101,10 @@ public class Config extends ConfigBase { }) public static boolean enable_java_udf = true; + @ConfField(description = "Whether to enable LOCAL TVF and FILE TVF with local storage. Disabled by default. " + + "Set this in fe.conf and restart every FE to enable it. Existing access checks still apply.") + public static boolean enable_local_tvf = false; + @ConfField(mutable = true, masterOnly = true, description = { "开启后，可以在导入时，利用创建的全局 java_udf 函数处理数据，默认为 false。", "When enabled, data can be processed using the globally created java_udf function during import." diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java index 297272cb582250..40959143298cdd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java @@ -241,6 +241,9 @@ protected void setScanParams(TFileRangeDesc rangeDesc, Split split) { if (!(split instanceof LanceSplit)) { throw new IllegalArgumentException("Expected LanceSplit but got " + split.getClass().getName()); } + // A bare local path has no URI scheme, so split inference cannot supply its type. + // Preserve the TVF's LOCAL/S3 identity for the BE's pre-open path validation. + rangeDesc.setFileType(tableValuedFunction.getTFileType()); LanceSplit lanceSplit = (LanceSplit) split; TLanceFileDesc lanceParams = new TLanceFileDesc(); lanceParams.setDatasetUri(lanceSplit.getDatasetUri()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java index 1d2580e52fc848..0fac5ba4344f39 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java @@ -116,6 +116,7 @@ import org.apache.doris.qe.AutoCloseSessionVariable; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; +import org.apache.doris.tablefunction.LocalTableValuedFunction; import org.apache.doris.thrift.TPartialUpdateNewRowPolicy; import com.google.common.base.Preconditions; @@ -659,6 +660,11 @@ private Plan bindTVFTableSink(MatchingContext> ctx) { "INSERT INTO TVF only supports local/s3/hdfs, but got: " + tvfName); } + // INSERT/EXPLAIN INSERT bind a sink without constructing the LOCAL read TVF. + if (tvfName.equals("local")) { + LocalTableValuedFunction.checkEnabled(); + } + // Validate required properties if (!properties.containsKey("file_path")) { throw new AnalysisException("TVF sink requires 'file_path' property"); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java index 4e4bf163145cf9..335ae199cc137a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java @@ -21,9 +21,12 @@ import org.apache.doris.analysis.StorageBackend.StorageType; import org.apache.doris.catalog.Env; import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.common.ErrorCode; import org.apache.doris.common.UserException; import org.apache.doris.datasource.FederationBackendPolicy; import org.apache.doris.datasource.property.storage.StorageProperties; +import org.apache.doris.mysql.privilege.PrivPredicate; import org.apache.doris.proto.InternalService; import org.apache.doris.proto.InternalService.PGlobResponse; import org.apache.doris.proto.InternalService.PGlobResponse.PFileInfo; @@ -61,6 +64,11 @@ public class LocalTableValuedFunction extends ExternalFileTableValuedFunction { private boolean sharedStorage = false; public LocalTableValuedFunction(Map properties) throws AnalysisException { + checkEnabled(); + + // Check privileges before listing files or discovering their schema. + checkAuth(ConnectContext.get()); + // 1. analyze common properties Map props = super.parseCommonProperties(properties); @@ -135,6 +143,24 @@ private void getFileListFromBackend() throws AnalysisException { } } + /** Check read and write admission before properties or backend access. */ + public static void checkEnabled() { + if (!Config.enable_local_tvf) { + throw new org.apache.doris.nereids.exceptions.AnalysisException( + "LOCAL TVF is disabled. Set enable_local_tvf=true in fe.conf and restart every FE to enable it."); + } + } + + @Override + public void checkAuth(ConnectContext ctx) { + if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN) + && !Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.OPERATOR)) { + String message = ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR.formatErrorMsg("ADMIN/OPERATOR"); + throw new org.apache.doris.nereids.exceptions.AnalysisException(message); + } + super.checkAuth(ctx); + } + @Override public TFileType getTFileType() { return TFileType.FILE_LOCAL; diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java index 91e79dda10e641..183e7304b6add2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java @@ -28,9 +28,17 @@ import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TFileType; +import com.google.common.collect.ImmutableSet; +import com.google.common.net.HostAndPort; +import com.google.common.net.InetAddresses; + +import java.net.URI; +import java.net.URISyntaxException; import java.util.Collections; import java.util.List; +import java.util.Locale; import java.util.Map; +import java.util.Set; /** * The Implement of table valued function @@ -46,6 +54,11 @@ public class S3TableValuedFunction extends ExternalFileTableValuedFunction { public static final String NAME = "s3"; + // Remote providers registered by the pinned Lance SDK. File providers also have aliases, + // so checking only for "file" (or a particular StorageProperties type) is insufficient. + private static final Set LANCE_REMOTE_SCHEMES = ImmutableSet.of( + "s3", "s3+ddb", "az", "abfss", "gs", "goosefs", "oss", "cos", "hf", "tos"); + public S3TableValuedFunction(Map properties) throws AnalysisException { this(properties, null); } @@ -63,6 +76,11 @@ public S3TableValuedFunction(Map properties, List properties, List '9') { + return false; + } + } + } + String host = parsed.getHost(); + // Rust rejects zones and IPv4 tails with leading zeros, such as ::ffff:127.0.0.01. + return !host.isEmpty() && (!host.contains(":") + || (!host.contains("%") && InetAddresses.isInetAddress(host))); + } catch (URISyntaxException | IllegalArgumentException e) { + return false; + } + } + // =========== implement abstract methods of ExternalFileTableValuedFunction ================= @Override public TFileType getTFileType() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java index f6db0c87ea05e6..aa3f1e726123ab 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/tvf/source/TVFScanNodeTest.java @@ -160,6 +160,7 @@ public void testLanceSplitsPinVersionAndFragments() throws Exception { ExternalFileTableValuedFunction tvf = Mockito.mock(ExternalFileTableValuedFunction.class); Mockito.when(table.getTvf()).thenReturn(tvf); Mockito.when(tvf.isLanceFormat()).thenReturn(true); + Mockito.when(tvf.getTFileType()).thenReturn(TFileType.FILE_S3); Mockito.when(tvf.getFilePath()).thenReturn("s3://bucket/table.lance"); Mockito.when(tvf.getLanceDatasetVersion()).thenReturn(42L); Mockito.when(tvf.getLanceFragments()).thenReturn(Arrays.asList( @@ -181,6 +182,7 @@ public void testLanceSplitsPinVersionAndFragments() throws Exception { TFileRangeDesc range = new TFileRangeDesc(); node.setScanParams(range, first); + Assert.assertEquals(TFileType.FILE_S3, range.getFileType()); Assert.assertTrue(range.isSetTableFormatParams()); Assert.assertEquals("lance", range.getTableFormatParams().getTableFormatType()); Assert.assertEquals("s3://bucket/table.lance", @@ -217,6 +219,27 @@ public void testLocalLanceUsesOneLatestWholeDatasetSplit() throws Exception { Assert.assertFalse(range.getTableFormatParams().getLanceParams().isSetFragmentIds()); } + @Test + public void testLocalLanceRangeRetainsFileTypeWithoutSchemeInference() throws Exception { + for (String path : Arrays.asList("data/table.lance", "/data/table.lance", "file:///data/table.lance")) { + SessionVariable sv = new SessionVariable(); + TupleDescriptor desc = new TupleDescriptor(new TupleId(0)); + FunctionGenTable table = Mockito.mock(FunctionGenTable.class); + ExternalFileTableValuedFunction tvf = Mockito.mock(ExternalFileTableValuedFunction.class); + Mockito.when(table.getTvf()).thenReturn(tvf); + Mockito.when(tvf.isLanceFormat()).thenReturn(true); + Mockito.when(tvf.getTFileType()).thenReturn(TFileType.FILE_LOCAL); + Mockito.when(tvf.getFilePath()).thenReturn(path); + desc.setTable(table); + TVFScanNode node = new TVFScanNode(new PlanNodeId(0), desc, false, sv, ScanContext.EMPTY); + TFileRangeDesc range = new TFileRangeDesc(); + node.setScanParams(range, node.getSplits(1).get(0)); + Assert.assertTrue(range.isSetFileType()); + Assert.assertEquals(TFileType.FILE_LOCAL, range.getFileType()); + Assert.assertEquals(path, range.getTableFormatParams().getLanceParams().getDatasetUri()); + } + } + // Verifies local Lance execution is restricted to the backend that provided its schema. @Test public void testLocalLancePinsExecutionToSchemaBackend() throws Exception { diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindTVFTableSinkTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindTVFTableSinkTest.java index b76ba57ddb5303..b789c4c391fd22 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindTVFTableSinkTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindTVFTableSinkTest.java @@ -17,25 +17,83 @@ package org.apache.doris.nereids.rules.analysis; +import org.apache.doris.common.Config; +import org.apache.doris.nereids.exceptions.AnalysisException; import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.trees.expressions.NamedExpression; import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.commands.ExplainCommand; import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTVFCommand; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; import org.apache.doris.nereids.trees.plans.logical.LogicalProject; import org.apache.doris.nereids.trees.plans.logical.LogicalTVFTableSink; import org.apache.doris.nereids.util.MemoTestUtils; import org.apache.doris.nereids.util.PlanChecker; import org.apache.doris.qe.ConnectContext; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.List; class BindTVFTableSinkTest { + private boolean originalEnableLocalTvf; + + @BeforeEach + void setUp() { + originalEnableLocalTvf = Config.enable_local_tvf; + Config.enable_local_tvf = false; + } + + @AfterEach + void tearDown() { + Config.enable_local_tvf = originalEnableLocalTvf; + } + + @Test + void disabledLocalSinkRejectsInsertAndExplainBeforeProperties() { + for (String prefix : Arrays.asList("", "EXPLAIN ")) { + for (String properties : Arrays.asList("'format'='csv'", + "'file_path'='/tmp/local_gate_', 'backend_id'='1', 'format'='csv'")) { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> analyzeInsert(prefix + "INSERT INTO local(" + properties + ") SELECT 1")); + Assertions.assertTrue(exception.getMessage().contains("LOCAL TVF is disabled")); + } + } + } + + @Test + void enabledLocalSinkKeepsPropertyValidation() { + Config.enable_local_tvf = true; + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> analyzeInsert("INSERT INTO local('format'='csv') SELECT 1")); + Assertions.assertTrue(exception.getMessage().contains("requires 'file_path'")); + } + + @Test + void disabledLocalDoesNotBlockRemoteSinks() { + for (String name : Arrays.asList("s3", "hdfs")) { + Assertions.assertInstanceOf(LogicalTVFTableSink.class, + analyzeInsert("INSERT INTO " + name + "('file_path'='remote/data', 'format'='csv') SELECT 1")); + } + } + + private Plan analyzeInsert(String sql) { + ConnectContext context = MemoTestUtils.createConnectContext(); + LogicalPlan parsed = new NereidsParser().parseSingle(sql); + if (parsed instanceof ExplainCommand) { + parsed = ((ExplainCommand) parsed).getLogicalPlan(); + } + InsertIntoTVFCommand command = (InsertIntoTVFCommand) parsed; + return PlanChecker.from(context).analyze(command.getExplainPlan(context)).getPlan(); + } @Test void duplicateOutputNamesKeepDistinctPositions() { + Config.enable_local_tvf = true; ConnectContext context = MemoTestUtils.createConnectContext(); InsertIntoTVFCommand command = (InsertIntoTVFCommand) new NereidsParser().parseSingle( "INSERT INTO local(" diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/LocalTableValuedFunctionEnableTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/LocalTableValuedFunctionEnableTest.java new file mode 100644 index 00000000000000..aaa96239628bf9 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/LocalTableValuedFunctionEnableTest.java @@ -0,0 +1,205 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.tablefunction; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Config; +import org.apache.doris.mysql.privilege.AccessControllerManager; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.proto.InternalService.PGlobResponse; +import org.apache.doris.proto.Types.PStatus; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.SessionVariable; +import org.apache.doris.resource.computegroup.ComputeGroup; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.system.Backend; +import org.apache.doris.system.SystemInfoService; +import org.apache.doris.thrift.TNetworkAddress; + +import com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +public class LocalTableValuedFunctionEnableTest { + private static final Map PROPERTIES = ImmutableMap.of( + "file_path", "local_privilege_test.csv", "backend_id", "1", "format", "csv"); + + private final Env env = Mockito.mock(Env.class); + private final ConnectContext context = Mockito.mock(ConnectContext.class); + private final AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class); + private final SystemInfoService systemInfo = Mockito.mock(SystemInfoService.class); + private final BackendServiceProxy proxy = Mockito.mock(BackendServiceProxy.class); + private boolean originalEnableLocalTvf; + private MockedStatic mockedEnv; + private MockedStatic mockedContext; + private MockedStatic mockedProxy; + + @BeforeEach + public void setUp() throws Exception { + originalEnableLocalTvf = Config.enable_local_tvf; + Config.enable_local_tvf = false; + mockedEnv = Mockito.mockStatic(Env.class); + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfo); + mockedContext = Mockito.mockStatic(ConnectContext.class); + mockedContext.when(ConnectContext::get).thenReturn(context); + mockedProxy = Mockito.mockStatic(BackendServiceProxy.class); + mockedProxy.when(BackendServiceProxy::getInstance).thenReturn(proxy); + Mockito.when(env.getAccessManager()).thenReturn(accessManager); + Mockito.when(context.getSessionVariable()).thenReturn(new SessionVariable()); + Mockito.when(context.getQueryTimeoutS()).thenReturn(60); + + Backend backend = Mockito.mock(Backend.class); + Mockito.when(backend.getId()).thenReturn(1L); + Mockito.when(backend.getHost()).thenReturn("127.0.0.1"); + Mockito.when(backend.isAlive()).thenReturn(true); + Mockito.when(backend.isQueryAvailable()).thenReturn(true); + Mockito.when(backend.isLoadAvailable()).thenReturn(true); + ComputeGroup computeGroup = Mockito.mock(ComputeGroup.class); + Mockito.when(context.getComputeGroup()).thenReturn(computeGroup); + Mockito.when(computeGroup.getBackendList()).thenReturn(Collections.singletonList(backend)); + Mockito.when(systemInfo.getBackend(1L)).thenReturn(backend); + Mockito.when(backend.getBrpcAddress()).thenReturn(new TNetworkAddress("127.0.0.1", 8060)); + Mockito.when(proxy.glob(Mockito.any(), Mockito.any())).thenReturn(CompletableFuture.completedFuture( + PGlobResponse.newBuilder().setStatus(PStatus.newBuilder().setStatusCode(0)).build())); + } + + @AfterEach + public void tearDown() { + Config.enable_local_tvf = originalEnableLocalTvf; + mockedProxy.close(); + mockedContext.close(); + mockedEnv.close(); + } + + @Test + public void testDisabledRejectsLocalBeforePropertyParsing() { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new LocalTableValuedFunction(ImmutableMap.of())); + Assertions.assertTrue(exception.getMessage().contains("enable_local_tvf=true")); + Mockito.verifyNoInteractions(accessManager, systemInfo, proxy); + } + + @Test + public void testDisabledRejectsCsvAndLanceBeforeFileAccess() { + for (String format : new String[] {"csv", "lance"}) { + Map properties = ImmutableMap.of( + "file_path", "local_enable_test.csv", "backend_id", "1", "format", format); + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new LocalTableValuedFunction(properties)); + Assertions.assertTrue(exception.getMessage().contains("LOCAL TVF is disabled")); + } + Mockito.verifyNoInteractions(accessManager, systemInfo, proxy); + } + + @Test + public void testEnabledLocalReachesExistingFileAccess() throws Exception { + Config.enable_local_tvf = true; + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true); + new LocalTableValuedFunction(PROPERTIES); + Mockito.verify(proxy).glob(Mockito.any(), Mockito.any()); + } + + @Test + public void testDisabledRejectsFileLocalBeforeFileAccess() { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new FileTableValuedFunction(PROPERTIES)); + Assertions.assertTrue(exception.getMessage().contains("LOCAL TVF is disabled")); + Mockito.verifyNoInteractions(accessManager, systemInfo, proxy); + } + + @Test + public void testEnabledFileLocalReachesExistingFileAccess() throws Exception { + Config.enable_local_tvf = true; + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true); + new FileTableValuedFunction(PROPERTIES); + Mockito.verify(proxy).glob(Mockito.any(), Mockito.any()); + } + + @Test + public void testDisabledLocalDoesNotBlockFileS3() { + Map properties = ImmutableMap.builder() + .put("uri", "s3://test-bucket/local_enable_test.csv") + .put("s3.endpoint", "s3.us-east-1.amazonaws.com") + .put("s3.region", "us-east-1") + .put("s3.access_key", "test-access-key") + .put("s3.secret_key", "test-secret-key") + .put("format", "csv") + .build(); + try (MockedConstruction mockedS3 = + Mockito.mockConstruction(S3TableValuedFunction.class)) { + Assertions.assertDoesNotThrow(() -> new FileTableValuedFunction(properties)); + Assertions.assertEquals(1, mockedS3.constructed().size()); + } + Mockito.verifyNoInteractions(accessManager, systemInfo, proxy); + } + + @Test + public void testEnabledLancePreservesExistingBackendSelection() throws Exception { + Config.enable_local_tvf = true; + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true); + new LocalTableValuedFunction(ImmutableMap.of( + "file_path", "local_enable_test.lance", "backend_id", "1", "format", "lance")); + Mockito.verifyNoInteractions(proxy); + } + + @Test + public void testEnabledLocalRejectsUnprivilegedUserBeforeFileAccess() { + Config.enable_local_tvf = true; + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new LocalTableValuedFunction(PROPERTIES)); + Assertions.assertTrue(exception.getMessage().contains("ADMIN/OPERATOR")); + Mockito.verifyNoInteractions(systemInfo, proxy); + } + + @Test + public void testOperatorCanAccessLocalFiles() throws Exception { + Config.enable_local_tvf = true; + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.OPERATOR)).thenReturn(true); + new LocalTableValuedFunction(PROPERTIES); + Mockito.verify(proxy).glob(Mockito.any(), Mockito.any()); + } + + @Test + public void testCheckAuthRejectsRevokedPrivileges() throws Exception { + Config.enable_local_tvf = true; + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true); + LocalTableValuedFunction local = new LocalTableValuedFunction(PROPERTIES); + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(false); + Assertions.assertThrows(AnalysisException.class, () -> local.checkAuth(context)); + } + + @Test + public void testEnabledFileRejectsUnprivilegedUserBeforeFileAccess() { + Config.enable_local_tvf = true; + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new FileTableValuedFunction(PROPERTIES)); + Assertions.assertTrue(exception.getMessage().contains("ADMIN/OPERATOR")); + Mockito.verifyNoInteractions(systemInfo, proxy); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java index 2d79b5a776a0de..6050a4209aa8b6 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java @@ -20,6 +20,7 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.FeConstants; import org.apache.doris.common.util.S3Util; +import org.apache.doris.datasource.lance.metadata.LanceMetadataLoader; import org.apache.doris.fs.FileSystemFactory; import org.apache.doris.nereids.StatementContext; import org.apache.doris.nereids.trees.expressions.Properties; @@ -42,6 +43,7 @@ import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -71,6 +73,109 @@ public void tearDown() { } } + @Test + public void testLanceRejectsLocalStorageBeforeOpeningDataset() { + org.apache.doris.common.AnalysisException exception = Assertions.assertThrows( + org.apache.doris.common.AnalysisException.class, + () -> new S3TableValuedFunction(ImmutableMap.of( + "file_path", "dataset.lance", "format", "lance"))); + Assertions.assertTrue(exception.getMessage().contains("requires remote storage")); + } + + @Test + public void testLanceRejectsBrokerLocalUrisBeforeOpeningDataset() { + FeConstants.runningUnitTest = false; + try (MockedStatic loader = Mockito.mockStatic(LanceMetadataLoader.class)) { + for (String uri : Arrays.asList("dataset.lance", "/tmp/dataset.lance", "../dataset.lance", + "//localhost/tmp/dataset.lance", "file:///tmp/dataset.lance", "file:/tmp/dataset.lance", + "file:dataset.lance", "FILE:///tmp/dataset.lance", "file://localhost/tmp/dataset.lance", + "file-object-store://ignored/tmp/dataset.lance", "file+uring://ignored/tmp/dataset.lance", + "FILE-OBJECT-STORE://ignored/tmp/dataset.lance", "FILE+URING://ignored/tmp/dataset.lance", + "c://ignored/tmp/dataset.lance", "memory://dataset.lance", "shared-memory://dataset.lance", + "s3://bucket:-0/../../dataset.lance", "s3://bucket:-00/../../dataset.lance", + "s3://bucket:+0/../../dataset.lance", "s3://[::ffff:127.0.0.01]/../../dataset.lance", + "s3://bucket:bad/../../dataset.lance", "s3://bucket:65536/../../dataset.lance", + "s3://:9000/dataset.lance", "s3://user@/dataset.lance", + "s3://[::1%25eth0]/dataset.lance", "https://bucket:bad/../../dataset.lance", + "", "s3:/bucket/dataset.lance", "s3://", "s3://bucket/bad path")) { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new S3TableValuedFunction(ImmutableMap.of( + "format", "lance", "uri", uri, "fs.broker.support", "true")), uri); + Assertions.assertTrue(exception.getMessage().contains("requires remote storage"), uri); + } + loader.verifyNoInteractions(); + } + } + + @Test + public void testLanceRejectsGuessedBrokerFileUrisBeforeOpeningDataset() { + FeConstants.runningUnitTest = false; + try (MockedStatic loader = Mockito.mockStatic(LanceMetadataLoader.class)) { + for (String uri : Arrays.asList("file:///tmp/dataset.lance", "file://localhost/tmp/dataset.lance")) { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new S3TableValuedFunction(ImmutableMap.of( + "format", "lance", "uri", uri, "broker.name", "test"))); + Assertions.assertTrue(exception.getMessage().contains("requires remote storage")); + } + loader.verifyNoInteractions(); + } + } + + @Test + public void testLanceRejectsBrokerWithoutUriBeforeOpeningDataset() { + FeConstants.runningUnitTest = false; + try (MockedStatic loader = Mockito.mockStatic(LanceMetadataLoader.class)) { + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new S3TableValuedFunction(ImmutableMap.of("format", "lance", "fs.broker.support", "true"))); + Assertions.assertTrue(exception.getMessage().contains("requires remote storage")); + loader.verifyNoInteractions(); + } + } + + @Test + public void testLanceRemoteUrisReachMetadataLoader() throws Exception { + FeConstants.runningUnitTest = false; + for (String uri : Arrays.asList("s3://bucket/dataset.lance", "s3+ddb://bucket/dataset.lance", + "oss://bucket/dataset.lance", "gs://bucket/dataset.lance", "az://container/dataset.lance", + "abfss://container@account.dfs.core.windows.net/dataset.lance", + "s3://bucket:0/dataset.lance", "s3://bucket:00080/dataset.lance", + "s3://bucket:65535/dataset.lance", "s3://bucket:/dataset.lance", + "s3://[::1]/dataset.lance", "s3://[::ffff:127.0.0.1]/dataset.lance", + "gs://bucket_with_underscore/dataset.lance", "s3://bucket:9000/dataset.lance", + "goosefs://bucket/dataset.lance", "goosefs:///dataset.lance", "goosefs:/dataset.lance", + "cos://bucket/dataset.lance", + "hf://datasets/repo/data.lance", + "tos://bucket/dataset.lance", "S3://bucket/dataset.lance")) { + assertLanceMetadataLoad(ImmutableMap.of( + "format", "lance", "uri", uri, "fs.broker.support", "true"), uri); + } + } + + @Test + public void testLanceHttpStorageUrisAreNormalizedBeforeValidation() throws Exception { + FeConstants.runningUnitTest = false; + Map s3Properties = new HashMap<>(PROPERTIES); + s3Properties.put("format", "lance"); + s3Properties.put("uri", "https://bucket.s3.us-west-2.amazonaws.com/dataset.lance"); + assertLanceMetadataLoad(s3Properties, "s3://bucket/dataset.lance"); + assertLanceMetadataLoad(ImmutableMap.of( + "format", "lance", "uri", "https://bucket.oss-cn-hangzhou.aliyuncs.com/dataset.lance", + "oss.endpoint", "oss-cn-hangzhou.aliyuncs.com", "oss.access_key", "ak", "oss.secret_key", "sk"), + "s3://bucket/dataset.lance"); + } + + private void assertLanceMetadataLoad(Map properties, String uri) throws Exception { + try (MockedStatic loader = Mockito.mockStatic(LanceMetadataLoader.class)) { + IllegalStateException reachedLoader = new IllegalStateException("remote metadata loader reached"); + loader.when(() -> LanceMetadataLoader.loadLatestForTvf(Mockito.eq(uri), Mockito.anyList())) + .thenThrow(reachedLoader); + AnalysisException exception = Assertions.assertThrows(AnalysisException.class, + () -> new S3TableValuedFunction(properties)); + Assertions.assertSame(reachedLoader, exception.getCause(), uri); + loader.verify(() -> LanceMetadataLoader.loadLatestForTvf(Mockito.eq(uri), Mockito.anyList())); + } + } + @Test public void testExactFilesKeepLiteralPathsAndSkipListing() throws Exception { FeConstants.runningUnitTest = false; diff --git a/regression-test/pipeline/cloud_p0/conf/fe_custom.conf b/regression-test/pipeline/cloud_p0/conf/fe_custom.conf index d6c2070aa939e2..7139bfa6e02aa8 100644 --- a/regression-test/pipeline/cloud_p0/conf/fe_custom.conf +++ b/regression-test/pipeline/cloud_p0/conf/fe_custom.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + stream_load_default_timeout_second = 72000 replication_num_forced_in_cloud_mode = true ignore_unsupported_properties_in_cloud_mode = true diff --git a/regression-test/pipeline/cloud_p1/conf/fe_custom.conf b/regression-test/pipeline/cloud_p1/conf/fe_custom.conf index 157f3a07a09b47..e671d0a7dd15b2 100644 --- a/regression-test/pipeline/cloud_p1/conf/fe_custom.conf +++ b/regression-test/pipeline/cloud_p1/conf/fe_custom.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + stream_load_default_timeout_second = 72000 replication_num_forced_in_cloud_mode = true ignore_unsupported_properties_in_cloud_mode = true diff --git a/regression-test/pipeline/external/conf/fe.conf b/regression-test/pipeline/external/conf/fe.conf index 4466998ae67163..c0160fa91f07b5 100644 --- a/regression-test/pipeline/external/conf/fe.conf +++ b/regression-test/pipeline/external/conf/fe.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information diff --git a/regression-test/pipeline/nonConcurrent/conf/fe.conf b/regression-test/pipeline/nonConcurrent/conf/fe.conf index e89a00d209ccf7..737e305a1bacf4 100644 --- a/regression-test/pipeline/nonConcurrent/conf/fe.conf +++ b/regression-test/pipeline/nonConcurrent/conf/fe.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information diff --git a/regression-test/pipeline/p0/conf/fe.conf b/regression-test/pipeline/p0/conf/fe.conf index 3d86c5c95e6332..494bc15cfb9bd1 100644 --- a/regression-test/pipeline/p0/conf/fe.conf +++ b/regression-test/pipeline/p0/conf/fe.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information diff --git a/regression-test/pipeline/p1/conf/fe.conf b/regression-test/pipeline/p1/conf/fe.conf index f20d51b62e6575..04037acd3644b9 100644 --- a/regression-test/pipeline/p1/conf/fe.conf +++ b/regression-test/pipeline/p1/conf/fe.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information diff --git a/regression-test/pipeline/vault_p0/conf/fe_custom.conf b/regression-test/pipeline/vault_p0/conf/fe_custom.conf index f62ffa19fc7676..f4c14df5d447b0 100644 --- a/regression-test/pipeline/vault_p0/conf/fe_custom.conf +++ b/regression-test/pipeline/vault_p0/conf/fe_custom.conf @@ -1,3 +1,6 @@ +# Exercise LOCAL TVF explicitly; production defaults to disabled. +enable_local_tvf=true + stream_load_default_timeout_second = 72000 replication_num_forced_in_cloud_mode = true ignore_unsupported_properties_in_cloud_mode = true diff --git a/regression-test/suites/account_p0/test_local_tvf_enable.groovy b/regression-test/suites/account_p0/test_local_tvf_enable.groovy new file mode 100644 index 00000000000000..3cc8bf71ce3d92 --- /dev/null +++ b/regression-test/suites/account_p0/test_local_tvf_enable.groovy @@ -0,0 +1,46 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_local_tvf_enable", "p0") { + def enabled = (sql "ADMIN SHOW FRONTEND CONFIG LIKE 'enable_local_tvf'")[0][1].toBoolean() + for (def functionName : ["LOCAL", "FILE"]) { + for (def prefix : ["", "EXPLAIN "]) { + test { + sql """${prefix}SELECT * FROM ${functionName}( + "file_path" = "local_enable_test.csv", + "backend_id" = "-1", + "format" = "csv" + )""" + // With opt-in enabled, the existing backend validation must be reached. + exception (enabled ? "'backend_id' is required" : "LOCAL TVF is disabled") + } + } + } + if (!enabled) { + for (def prefix : ["", "EXPLAIN "]) { + test { + sql """${prefix}INSERT INTO LOCAL( + "file_path" = "/tmp/local_enable_test_", + "backend_id" = "-1", + "format" = "csv" + ) SELECT 1""" + exception "LOCAL TVF is disabled" + } + } + } + +}