Skip to content

Commit 45cbf17

Browse files
wgtmacwgtmac
committed
polish impl
1 parent 1734fe0 commit 45cbf17

File tree

12 files changed

+75
-134
lines changed

12 files changed

+75
-134
lines changed

‎src/iceberg/catalog.h‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -179,8 +179,7 @@ class ICEBERG_EXPORT Catalog {
179179
/// \param identifier a table identifier
180180
/// \return instance of Table implementation referred to by identifier or
181181
/// ErrorKind::kNoSuchTable if the table does not exist
182-
virtual Result<std::shared_ptr<Table>> LoadTable(
183-
const TableIdentifier& identifier) const = 0;
182+
virtual Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) = 0;
184183

185184
/// \brief Register a table with the catalog if it does not exist
186185
///

‎src/iceberg/catalog/memory/in_memory_catalog.cc‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -519,7 +519,7 @@ Status InMemoryCatalog::RenameTable(const TableIdentifier& from,
519519
}
520520

521521
Result<std::shared_ptr<Table>> InMemoryCatalog::LoadTable(
522-
const TableIdentifier& identifier) const {
522+
const TableIdentifier& identifier) {
523523
if (!file_io_) [[unlikely]] {
524524
return InvalidArgument("file_io is not set for catalog {}", catalog_name_);
525525
}
@@ -533,9 +533,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::LoadTable(
533533

534534
ICEBERG_ASSIGN_OR_RAISE(auto metadata,
535535
TableMetadataUtil::Read(*file_io_, metadata_location));
536-
auto non_const_catalog = std::const_pointer_cast<InMemoryCatalog>(shared_from_this());
537536
return Table::Make(identifier, std::move(metadata), std::move(metadata_location),
538-
file_io_, non_const_catalog);
537+
file_io_, shared_from_this());
539538
}
540539

541540
Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(

‎src/iceberg/catalog/memory/in_memory_catalog.h‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,8 +93,7 @@ class ICEBERG_EXPORT InMemoryCatalog
9393

9494
Status RenameTable(const TableIdentifier& from, const TableIdentifier& to) override;
9595

96-
Result<std::shared_ptr<Table>> LoadTable(
97-
const TableIdentifier& identifier) const override;
96+
Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
9897

9998
Result<std::shared_ptr<Table>> RegisterTable(
10099
const TableIdentifier& identifier,

‎src/iceberg/catalog/rest/http_client.cc‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -136,11 +136,11 @@ Status HandleFailureResponse(const cpr::Response& response,
136136

137137
void HttpClient::PrepareSession(
138138
const std::string& path, const std::unordered_map<std::string, std::string>& params,
139-
const std::unordered_map<std::string, std::string>& request_headers) {
139+
const std::unordered_map<std::string, std::string>& headers) {
140140
session_->SetUrl(cpr::Url{path});
141141
session_->SetParameters(GetParameters(params));
142142
session_->RemoveContent();
143-
auto final_headers = MergeHeaders(default_headers_, request_headers);
143+
auto final_headers = MergeHeaders(default_headers_, headers);
144144
session_->SetHeader(final_headers);
145145
}
146146

‎src/iceberg/catalog/rest/http_client.h‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -109,9 +109,9 @@ class ICEBERG_REST_EXPORT HttpClient {
109109
const ErrorHandler& error_handler);
110110

111111
private:
112-
void PrepareSession(
113-
const std::string& path, const std::unordered_map<std::string, std::string>& params,
114-
const std::unordered_map<std::string, std::string>& request_headers);
112+
void PrepareSession(const std::string& path,
113+
const std::unordered_map<std::string, std::string>& params,
114+
const std::unordered_map<std::string, std::string>& headers);
115115

116116
std::unordered_map<std::string, std::string> default_headers_;
117117

‎src/iceberg/catalog/rest/rest_catalog.cc‎

Lines changed: 56 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,33 @@ Result<CatalogConfig> FetchServerConfig(const ResourcePaths& paths,
7474
return CatalogConfigFromJson(json);
7575
}
7676

77+
#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \
78+
do { \
79+
if (!endpoints.contains(endpoint)) { \
80+
return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \
81+
} \
82+
} while (0)
83+
84+
Result<bool> CaptureNoSuchObject(const auto& status, ErrorKind kind) {
85+
ICEBERG_DCHECK(kind == ErrorKind::kNoSuchTable || kind == ErrorKind::kNoSuchNamespace,
86+
"Invalid kind for CaptureNoSuchObject");
87+
if (status.has_value()) {
88+
return true;
89+
}
90+
if (status.error().kind == kind) {
91+
return false;
92+
}
93+
return std::unexpected(status.error());
94+
}
95+
96+
Result<bool> CaptureNoSuchTable(const auto& status) {
97+
return CaptureNoSuchObject(status, ErrorKind::kNoSuchTable);
98+
}
99+
100+
Result<bool> CaptureNoSuchNamespace(const auto& status) {
101+
return CaptureNoSuchObject(status, ErrorKind::kNoSuchNamespace);
102+
}
103+
77104
} // namespace
78105

79106
RestCatalog::~RestCatalog() = default;
@@ -126,9 +153,7 @@ RestCatalog::RestCatalog(std::unique_ptr<RestCatalogProperties> config,
126153
std::string_view RestCatalog::name() const { return name_; }
127154

128155
Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns) const {
129-
ICEBERG_RETURN_UNEXPECTED(
130-
CheckEndpoint(supported_endpoints_, Endpoint::ListNamespaces()));
131-
156+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::ListNamespaces());
132157
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespaces());
133158
std::vector<Namespace> result;
134159
std::string next_token;
@@ -157,9 +182,7 @@ Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns)
157182

158183
Status RestCatalog::CreateNamespace(
159184
const Namespace& ns, const std::unordered_map<std::string, std::string>& properties) {
160-
ICEBERG_RETURN_UNEXPECTED(
161-
CheckEndpoint(supported_endpoints_, Endpoint::CreateNamespace()));
162-
185+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CreateNamespace());
163186
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespaces());
164187
CreateNamespaceRequest request{.namespace_ = ns, .properties = properties};
165188
ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
@@ -173,9 +196,7 @@ Status RestCatalog::CreateNamespace(
173196

174197
Result<std::unordered_map<std::string, std::string>> RestCatalog::GetNamespaceProperties(
175198
const Namespace& ns) const {
176-
ICEBERG_RETURN_UNEXPECTED(
177-
CheckEndpoint(supported_endpoints_, Endpoint::GetNamespaceProperties()));
178-
199+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::GetNamespaceProperties());
179200
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
180201
ICEBERG_ASSIGN_OR_RAISE(const auto response,
181202
client_->Get(path, /*params=*/{}, /*headers=*/{},
@@ -186,48 +207,29 @@ Result<std::unordered_map<std::string, std::string>> RestCatalog::GetNamespacePr
186207
}
187208

188209
Status RestCatalog::DropNamespace(const Namespace& ns) {
189-
ICEBERG_RETURN_UNEXPECTED(
190-
CheckEndpoint(supported_endpoints_, Endpoint::DropNamespace()));
210+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::DropNamespace());
191211
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
192-
193212
ICEBERG_ASSIGN_OR_RAISE(const auto response,
194213
client_->Delete(path, /*params=*/{}, /*headers=*/{},
195214
*DropNamespaceErrorHandler::Instance()));
196215
return {};
197216
}
198217

199218
Result<bool> RestCatalog::NamespaceExists(const Namespace& ns) const {
200-
auto check = CheckEndpoint(supported_endpoints_, Endpoint::NamespaceExists());
201-
if (!check.has_value()) {
219+
if (!supported_endpoints_.contains(Endpoint::NamespaceExists())) {
202220
// Fall back to GetNamespaceProperties
203-
auto result = GetNamespaceProperties(ns);
204-
if (!result.has_value() && result.error().kind == ErrorKind::kNoSuchNamespace) {
205-
return false;
206-
}
207-
ICEBERG_RETURN_UNEXPECTED(result);
208-
// GetNamespaceProperties succeeded, namespace exists
209-
return true;
221+
return CaptureNoSuchNamespace(GetNamespaceProperties(ns));
210222
}
211223

212224
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
213-
auto response_or_error =
214-
client_->Head(path, /*headers=*/{}, *NamespaceErrorHandler::Instance());
215-
if (!response_or_error.has_value()) {
216-
// catch NoSuchNamespaceException/404 and return false
217-
if (response_or_error.error().kind == ErrorKind::kNoSuchNamespace) {
218-
return false;
219-
}
220-
ICEBERG_RETURN_UNEXPECTED(response_or_error);
221-
}
222-
return true;
225+
return CaptureNoSuchNamespace(
226+
client_->Head(path, /*headers=*/{}, *NamespaceErrorHandler::Instance()));
223227
}
224228

225229
Status RestCatalog::UpdateNamespaceProperties(
226230
const Namespace& ns, const std::unordered_map<std::string, std::string>& updates,
227231
const std::unordered_set<std::string>& removals) {
228-
ICEBERG_RETURN_UNEXPECTED(
229-
CheckEndpoint(supported_endpoints_, Endpoint::UpdateNamespace()));
230-
232+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::UpdateNamespace());
231233
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->NamespaceProperties(ns));
232234
UpdateNamespacePropertiesRequest request{
233235
.removals = std::vector<std::string>(removals.begin(), removals.end()),
@@ -252,7 +254,7 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
252254
const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
253255
const std::string& location,
254256
const std::unordered_map<std::string, std::string>& properties) {
255-
ICEBERG_RETURN_UNEXPECTED(CheckEndpoint(supported_endpoints_, Endpoint::CreateTable()));
257+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CreateTable());
256258
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(identifier.ns));
257259

258260
CreateTableRequest request{
@@ -295,7 +297,7 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
295297
}
296298

297299
Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {
298-
ICEBERG_RETURN_UNEXPECTED(CheckEndpoint(supported_endpoints_, Endpoint::DeleteTable()));
300+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::DeleteTable());
299301
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
300302

301303
std::unordered_map<std::string, std::string> params;
@@ -309,53 +311,42 @@ Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {
309311
}
310312

311313
Result<bool> RestCatalog::TableExists(const TableIdentifier& identifier) const {
312-
auto check = CheckEndpoint(supported_endpoints_, Endpoint::TableExists());
313-
if (!check.has_value()) {
314-
// Fall back to LoadTable endpoint (GET)
315-
auto result = LoadTable(identifier);
316-
if (!result.has_value() && result.error().kind == ErrorKind::kNoSuchTable) {
317-
return false;
318-
}
319-
ICEBERG_RETURN_UNEXPECTED(result);
320-
// LoadTable succeeded, table exists
321-
return true;
314+
if (!supported_endpoints_.contains(Endpoint::TableExists())) {
315+
// Fall back to call LoadTable
316+
return CaptureNoSuchTable(LoadTableInternal(identifier));
322317
}
323318

324319
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
325-
auto response_or_error =
326-
client_->Head(path, /*headers=*/{}, *TableErrorHandler::Instance());
327-
if (!response_or_error.has_value()) {
328-
// catch NoSuchTableException/404 and return false
329-
if (response_or_error.error().kind == ErrorKind::kNoSuchTable) {
330-
return false;
331-
}
332-
ICEBERG_RETURN_UNEXPECTED(response_or_error);
333-
}
334-
return true;
320+
return CaptureNoSuchTable(
321+
client_->Head(path, /*headers=*/{}, *TableErrorHandler::Instance()));
335322
}
336323

337324
Status RestCatalog::RenameTable([[maybe_unused]] const TableIdentifier& from,
338325
[[maybe_unused]] const TableIdentifier& to) {
339326
return NotImplemented("Not implemented");
340327
}
341328

342-
Result<std::shared_ptr<Table>> RestCatalog::LoadTable(
329+
Result<std::string> RestCatalog::LoadTableInternal(
343330
const TableIdentifier& identifier) const {
344-
ICEBERG_RETURN_UNEXPECTED(CheckEndpoint(supported_endpoints_, Endpoint::LoadTable()));
331+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::LoadTable());
345332
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
346-
347333
ICEBERG_ASSIGN_OR_RAISE(
348334
const auto response,
349335
client_->Get(path, /*params=*/{}, /*headers=*/{}, *TableErrorHandler::Instance()));
336+
return response.body();
337+
}
350338

351-
// TODO(Feiyang Li): support load metadata table
352-
ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
339+
Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
340+
ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::LoadTable());
341+
ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
342+
343+
ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
344+
ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
353345
ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
354-
// Cast away const since Table needs non-const Catalog pointer for mutations
355-
auto non_const_catalog = std::const_pointer_cast<RestCatalog>(shared_from_this());
356-
return Table::Make(identifier, load_result.metadata,
346+
347+
return Table::Make(identifier, std::move(load_result.metadata),
357348
std::move(load_result.metadata_location), file_io_,
358-
non_const_catalog);
349+
shared_from_this());
359350
}
360351

361352
Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(

‎src/iceberg/catalog/rest/rest_catalog.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -97,8 +97,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
9797

9898
Status DropTable(const TableIdentifier& identifier, bool purge) override;
9999

100-
Result<std::shared_ptr<Table>> LoadTable(
101-
const TableIdentifier& identifier) const override;
100+
Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
102101

103102
Result<std::shared_ptr<Table>> RegisterTable(
104103
const TableIdentifier& identifier,
@@ -109,6 +108,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
109108
std::shared_ptr<FileIO> file_io, std::unique_ptr<ResourcePaths> paths,
110109
std::unordered_set<Endpoint> endpoints);
111110

111+
Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
112+
112113
std::unique_ptr<RestCatalogProperties> config_;
113114
std::shared_ptr<FileIO> file_io_;
114115
std::unique_ptr<HttpClient> client_;

‎src/iceberg/catalog/rest/rest_util.cc‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -253,12 +253,4 @@ std::string GetStandardReasonPhrase(int32_t status_code) {
253253
}
254254
}
255255

256-
Status CheckEndpoint(const std::unordered_set<Endpoint>& supported_endpoints,
257-
const Endpoint& endpoint) {
258-
if (!supported_endpoints.contains(endpoint)) {
259-
return NotSupported("Server does not support endpoint: {}", endpoint.ToString());
260-
}
261-
return {};
262-
}
263-
264256
} // namespace iceberg::rest

‎src/iceberg/catalog/rest/rest_util.h‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -93,12 +93,4 @@ ICEBERG_REST_EXPORT std::unordered_map<std::string, std::string> MergeConfigs(
9393
/// Error").
9494
ICEBERG_REST_EXPORT std::string GetStandardReasonPhrase(int32_t status_code);
9595

96-
/// \brief Check whether the given endpoint is in the set of supported endpoints.
97-
///
98-
/// \param supported_endpoints Set of endpoints advertised by the server
99-
/// \param endpoint Endpoint to validate
100-
/// \return Status::OK if supported, NotSupported error otherwise
101-
ICEBERG_REST_EXPORT Status CheckEndpoint(
102-
const std::unordered_set<Endpoint>& supported_endpoints, const Endpoint& endpoint);
103-
10496
} // namespace iceberg::rest

‎src/iceberg/test/mock_catalog.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ class MockCatalog : public Catalog {
8181
(override));
8282

8383
MOCK_METHOD((Result<std::shared_ptr<Table>>), LoadTable, (const TableIdentifier&),
84-
(const, override));
84+
(override));
8585

8686
MOCK_METHOD((Result<std::shared_ptr<Table>>), RegisterTable,
8787
(const TableIdentifier&, const std::string&), (override));

0 commit comments

Comments
 (0)