Skip to content

Commit 29205ff

Browse files
committed
polish design and refine impl
1 parent e9a2878 commit 29205ff

20 files changed

Lines changed: 525 additions & 520 deletions

‎src/iceberg/arrow/arrow_io.cc‎

Lines changed: 26 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
#include <limits>
2323
#include <mutex>
2424
#include <optional>
25-
#include <unordered_map>
2625
#include <vector>
2726

2827
#include <arrow/buffer.h>
@@ -36,8 +35,6 @@
3635
#include "iceberg/arrow/arrow_io_internal.h"
3736
#include "iceberg/arrow/arrow_io_util.h"
3837
#include "iceberg/arrow/arrow_status_internal.h"
39-
#include "iceberg/arrow/s3/arrow_s3_internal.h"
40-
#include "iceberg/util/location_util.h"
4138
#include "iceberg/util/macros.h"
4239

4340
namespace iceberg::arrow {
@@ -486,53 +483,13 @@ class ArrowOutputFile : public OutputFile {
486483

487484
} // namespace
488485

489-
const std::shared_ptr<::arrow::fs::FileSystem>& ArrowFileSystemFileIO::FileSystemForPath(
490-
std::string_view location) const {
491-
if (fs_by_prefix_.empty()) {
492-
return arrow_fs_;
493-
}
494-
// Longest matching prefix wins; fall back to the default file system.
495-
const std::string canonical = LocationUtil::CanonicalizeS3Scheme(location);
496-
const std::shared_ptr<::arrow::fs::FileSystem>* best = &arrow_fs_;
497-
size_t best_len = 0;
498-
for (const auto& [prefix, fs] : fs_by_prefix_) {
499-
if (prefix.size() > best_len && LocationUtil::PathHasPrefix(canonical, prefix)) {
500-
best = &fs;
501-
best_len = prefix.size();
502-
}
503-
}
504-
return *best;
505-
}
506-
507-
Status ArrowFileSystemFileIO::SetStorageCredentials(
508-
const std::vector<
509-
std::pair<std::string, std::unordered_map<std::string, std::string>>>&
510-
properties_by_prefix) {
511-
#if ICEBERG_S3_ENABLED
512-
std::vector<std::pair<std::string, std::shared_ptr<::arrow::fs::FileSystem>>>
513-
fs_by_prefix;
514-
fs_by_prefix.reserve(properties_by_prefix.size());
515-
for (const auto& [prefix, properties] : properties_by_prefix) {
516-
ICEBERG_ASSIGN_OR_RAISE(auto fs, BuildArrowS3FileSystem(properties));
517-
fs_by_prefix.emplace_back(prefix, std::move(fs));
518-
}
519-
fs_by_prefix_ = std::move(fs_by_prefix);
520-
return {};
521-
#else
522-
(void)properties_by_prefix;
523-
return NotSupported("S3 storage credentials require Arrow S3 support");
524-
#endif
525-
}
526-
527-
Result<std::string> ArrowFileSystemFileIO::ResolvePath(
528-
const std::shared_ptr<::arrow::fs::FileSystem>& fs,
529-
const std::string& file_location) {
486+
Result<std::string> ArrowFileSystemFileIO::ResolvePath(const std::string& file_location) {
530487
const auto pos = file_location.find("://");
531488
if (pos == std::string::npos) {
532489
return file_location;
533490
}
534491

535-
auto path = fs->PathFromUri(file_location);
492+
auto path = arrow_fs_->PathFromUri(file_location);
536493
if (path.ok()) {
537494
return std::move(path).ValueOrDie();
538495
}
@@ -555,14 +512,14 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenArrowInputStream(
555512
ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
556513

557514
if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
558-
const auto& fs = arrow_io->FileSystemForPath(path);
559-
ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(fs, path));
515+
ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
560516
::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File);
561517
if (length.has_value()) {
562518
ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length));
563519
file_info.set_size(size);
564520
}
565-
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, fs->OpenInputFile(file_info));
521+
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input,
522+
arrow_io->arrow_fs_->OpenInputFile(file_info));
566523
return input;
567524
}
568525

@@ -586,15 +543,16 @@ Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
586543
ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
587544

588545
if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
589-
const auto& fs = arrow_io->FileSystemForPath(path);
590-
ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(fs, path));
546+
ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
591547
if (!overwrite) {
592-
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto info, fs->GetFileInfo(resolved_path));
548+
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto info,
549+
arrow_io->arrow_fs_->GetFileInfo(resolved_path));
593550
if (info.type() != ::arrow::fs::FileType::NotFound) {
594551
return AlreadyExists("File already exists: {}", path);
595552
}
596553
}
597-
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto output, fs->OpenOutputStream(resolved_path));
554+
ICEBERG_ARROW_ASSIGN_OR_RETURN(auto output,
555+
arrow_io->arrow_fs_->OpenOutputStream(resolved_path));
598556
return output;
599557
}
600558

@@ -610,60 +568,42 @@ Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
610568

611569
Result<std::unique_ptr<InputFile>> ArrowFileSystemFileIO::NewInputFile(
612570
std::string file_location) {
613-
const auto& fs = FileSystemForPath(file_location);
614-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
615-
return std::make_unique<ArrowInputFile>(fs, std::move(file_location), std::move(path),
616-
std::nullopt);
571+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
572+
return std::make_unique<ArrowInputFile>(arrow_fs_, std::move(file_location),
573+
std::move(path), std::nullopt);
617574
}
618575

619576
Result<std::unique_ptr<InputFile>> ArrowFileSystemFileIO::NewInputFile(
620577
std::string file_location, size_t length) {
621578
ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(length));
622-
const auto& fs = FileSystemForPath(file_location);
623-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
624-
return std::make_unique<ArrowInputFile>(fs, std::move(file_location), std::move(path),
625-
size);
579+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
580+
return std::make_unique<ArrowInputFile>(arrow_fs_, std::move(file_location),
581+
std::move(path), size);
626582
}
627583

628584
Result<std::unique_ptr<OutputFile>> ArrowFileSystemFileIO::NewOutputFile(
629585
std::string file_location) {
630-
const auto& fs = FileSystemForPath(file_location);
631-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
632-
return std::make_unique<ArrowOutputFile>(fs, std::move(file_location), std::move(path));
586+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
587+
return std::make_unique<ArrowOutputFile>(arrow_fs_, std::move(file_location),
588+
std::move(path));
633589
}
634590

635591
/// \brief Delete a file at the given location.
636592
Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
637-
const auto& fs = FileSystemForPath(file_location);
638-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
639-
ICEBERG_ARROW_RETURN_NOT_OK(fs->DeleteFile(path));
593+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
594+
ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(path));
640595
return {};
641596
}
642597

643598
Status ArrowFileSystemFileIO::DeleteFiles(
644599
const std::vector<std::string>& file_locations) {
645-
if (fs_by_prefix_.empty()) {
646-
// No per-prefix routing: one batched delete on the default file system.
647-
std::vector<std::string> paths;
648-
paths.reserve(file_locations.size());
649-
for (const auto& file_location : file_locations) {
650-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(arrow_fs_, file_location));
651-
paths.push_back(std::move(path));
652-
}
653-
ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
654-
return {};
655-
}
656-
657-
// Paths may route to different file systems: group by fs, then batch per fs.
658-
std::unordered_map<::arrow::fs::FileSystem*, std::vector<std::string>> paths_by_fs;
600+
std::vector<std::string> paths;
601+
paths.reserve(file_locations.size());
659602
for (const auto& file_location : file_locations) {
660-
const auto& fs = FileSystemForPath(file_location);
661-
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(fs, file_location));
662-
paths_by_fs[fs.get()].push_back(std::move(path));
663-
}
664-
for (auto& [fs, paths] : paths_by_fs) {
665-
ICEBERG_ARROW_RETURN_NOT_OK(fs->DeleteFiles(paths));
603+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
604+
paths.push_back(std::move(path));
666605
}
606+
ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
667607
return {};
668608
}
669609

‎src/iceberg/arrow/arrow_io_internal.h‎

Lines changed: 3 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,6 @@
2323
#include <memory>
2424
#include <optional>
2525
#include <string>
26-
#include <string_view>
27-
#include <unordered_map>
28-
#include <utility>
2926
#include <vector>
3027

3128
#include <arrow/filesystem/type_fwd.h>
@@ -55,8 +52,7 @@ OpenArrowOutputStream(const std::shared_ptr<FileIO>& io, const std::string& path
5552
bool overwrite = true);
5653

5754
/// \brief A concrete implementation of FileIO for Arrow file system.
58-
class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO,
59-
public SupportsStorageCredentials {
55+
class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
6056
public:
6157
explicit ArrowFileSystemFileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs)
6258
: arrow_fs_(std::move(arrow_fs)) {}
@@ -85,12 +81,6 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO,
8581
/// \brief Delete files at the given locations.
8682
Status DeleteFiles(const std::vector<std::string>& file_locations) override;
8783

88-
/// \brief Build one S3 file system per credential prefix for per-path routing.
89-
Status SetStorageCredentials(
90-
const std::vector<
91-
std::pair<std::string, std::unordered_map<std::string, std::string>>>&
92-
properties_by_prefix) override;
93-
9484
/// \brief Get the Arrow file system.
9585
const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
9686

@@ -102,18 +92,10 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO,
10292
friend Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
10393
const std::shared_ptr<FileIO>& io, const std::string& path, bool overwrite);
10494

105-
/// \brief Pick the file system for `location` (longest matching prefix, else
106-
/// the default).
107-
const std::shared_ptr<::arrow::fs::FileSystem>& FileSystemForPath(
108-
std::string_view location) const;
109-
110-
/// \brief Resolve a file location to a filesystem path using `fs`.
111-
Result<std::string> ResolvePath(const std::shared_ptr<::arrow::fs::FileSystem>& fs,
112-
const std::string& file_location);
95+
/// \brief Resolve a file location to a filesystem path.
96+
Result<std::string> ResolvePath(const std::string& file_location);
11397

11498
std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
115-
std::vector<std::pair<std::string, std::shared_ptr<::arrow::fs::FileSystem>>>
116-
fs_by_prefix_;
11799
};
118100

119101
} // namespace iceberg::arrow

0 commit comments

Comments
 (0)