Skip to content

Commit dbf9592

Browse files
zhjwpkuzhjwpku
authored
chore: use chrono::milliseconds in snapshot and consolidate error usage (#83)
reduce direct unexpected usage by using more Error wrappers defined in result.h --------- Signed-off-by: Junwang Zhao <[email protected]>
1 parent 2aed97b commit dbf9592

20 files changed

+232
-242
lines changed

‎src/iceberg/CMakeLists.txt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,8 @@ set(ICEBERG_SOURCES
3636
transform_function.cc
3737
type.cc
3838
snapshot.cc
39-
util/murmurhash3_internal.cc)
39+
util/murmurhash3_internal.cc
40+
util/timepoint.cc)
4041

4142
set(ICEBERG_STATIC_BUILD_INTERFACE_LIBS)
4243
set(ICEBERG_SHARED_BUILD_INTERFACE_LIBS)

‎src/iceberg/expression/expression.cc‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ std::string And::ToString() const {
5151

5252
Result<std::shared_ptr<Expression>> And::Negate() const {
5353
// TODO(yingcai-cy): Implement Or expression
54-
return InvalidExpressionError("And negation not yet implemented");
54+
return InvalidExpression("And negation not yet implemented");
5555
}
5656

5757
bool And::Equals(const Expression& expr) const {

‎src/iceberg/expression/expression.h‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,8 +68,7 @@ class ICEBERG_EXPORT Expression {
6868

6969
/// \brief Returns the negation of this expression, equivalent to not(this).
7070
virtual Result<std::shared_ptr<Expression>> Negate() const {
71-
return unexpected(
72-
Error(ErrorKind::kInvalidExpression, "Expression cannot be negated"));
71+
return InvalidExpression("Expression cannot be negated");
7372
}
7473

7574
/// \brief Returns whether this expression will accept the same values as another.

‎src/iceberg/file_io.h‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,7 @@ class ICEBERG_EXPORT FileIO {
5252
virtual Result<std::string> ReadFile(const std::string& file_location,
5353
std::optional<size_t> length) {
5454
// We provide a default implementation to avoid Windows linker error LNK2019.
55-
return unexpected<Error>{
56-
{.kind = ErrorKind::kNotImplemented, .message = "ReadFile not implemented"}};
55+
return NotImplemented("ReadFile not implemented");
5756
}
5857

5958
/// \brief Write the given content to the file at the given location.
@@ -64,17 +63,15 @@ class ICEBERG_EXPORT FileIO {
6463
/// file exists.
6564
/// \return void if the write succeeded, an error code if the write failed.
6665
virtual Status WriteFile(const std::string& file_location, std::string_view content) {
67-
return unexpected<Error>{
68-
{.kind = ErrorKind::kNotImplemented, .message = "WriteFile not implemented"}};
66+
return NotImplemented("WriteFile not implemented");
6967
}
7068

7169
/// \brief Delete a file at the given location.
7270
///
7371
/// \param file_location The location of the file to delete.
7472
/// \return void if the delete succeeded, an error code if the delete failed.
7573
virtual Status DeleteFile(const std::string& file_location) {
76-
return unexpected<Error>{
77-
{.kind = ErrorKind::kNotImplemented, .message = "DeleteFile not implemented"}};
74+
return NotImplemented("DeleteFile not implemented");
7875
}
7976
};
8077

‎src/iceberg/json_internal.cc‎

Lines changed: 9 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@
2727
#include <type_traits>
2828
#include <unordered_set>
2929

30-
#include <iceberg/table.h>
3130
#include <nlohmann/json.hpp>
3231

3332
#include "iceberg/partition_field.h"
@@ -38,11 +37,13 @@
3837
#include "iceberg/snapshot.h"
3938
#include "iceberg/sort_order.h"
4039
#include "iceberg/statistics_file.h"
40+
#include "iceberg/table.h"
4141
#include "iceberg/table_metadata.h"
4242
#include "iceberg/transform.h"
4343
#include "iceberg/type.h"
4444
#include "iceberg/util/formatter.h" // IWYU pragma: keep
4545
#include "iceberg/util/macros.h"
46+
#include "iceberg/util/timepoint.h"
4647

4748
namespace iceberg {
4849

@@ -502,7 +503,7 @@ nlohmann::json ToJson(const Snapshot& snapshot) {
502503
if (snapshot.sequence_number > TableMetadata::kInitialSequenceNumber) {
503504
json[kSequenceNumber] = snapshot.sequence_number;
504505
}
505-
json[kTimestampMs] = snapshot.timestamp_ms;
506+
json[kTimestampMs] = UnixMsFromTimePointMs(snapshot.timestamp_ms);
506507
json[kManifestList] = snapshot.manifest_list;
507508
// If there is an operation, write the summary map
508509
if (snapshot.operation().has_value()) {
@@ -722,7 +723,9 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
722723
ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
723724
ICEBERG_ASSIGN_OR_RAISE(auto sequence_number,
724725
GetJsonValueOptional<int64_t>(json, kSequenceNumber));
725-
ICEBERG_ASSIGN_OR_RAISE(auto timestamp_ms, GetJsonValue<int64_t>(json, kTimestampMs));
726+
ICEBERG_ASSIGN_OR_RAISE(
727+
auto timestamp_ms,
728+
GetJsonValue<int64_t>(json, kTimestampMs).and_then(TimePointMsFromUnixMs));
726729
ICEBERG_ASSIGN_OR_RAISE(auto manifest_list,
727730
GetJsonValue<std::string>(json, kManifestList));
728731

@@ -735,24 +738,14 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
735738
if (summary_json.has_value()) {
736739
for (const auto& [key, value] : summary_json->items()) {
737740
if (!kValidSnapshotSummaryFields.contains(key)) {
738-
return unexpected<Error>({
739-
.kind = ErrorKind::kJsonParseError,
740-
.message = std::format("Invalid snapshot summary field: {}", key),
741-
});
741+
return JsonParseError("Invalid snapshot summary field: {}", key);
742742
}
743743
if (!value.is_string()) {
744-
return unexpected<Error>({
745-
.kind = ErrorKind::kJsonParseError,
746-
.message =
747-
std::format("Invalid snapshot summary field value: {}", value.dump()),
748-
});
744+
return JsonParseError("Invalid snapshot summary field value: {}", value.dump());
749745
}
750746
if (key == SnapshotSummaryFields::kOperation &&
751747
!kValidDataOperation.contains(value.get<std::string>())) {
752-
return unexpected<Error>({
753-
.kind = ErrorKind::kJsonParseError,
754-
.message = std::format("Invalid snapshot operation: {}", value.dump()),
755-
});
748+
return JsonParseError("Invalid snapshot operation: {}", value.dump());
756749
}
757750
summary[key] = value.get<std::string>();
758751
}

‎src/iceberg/result.h‎

Lines changed: 29 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -28,21 +28,20 @@
2828
namespace iceberg {
2929

3030
/// \brief Error types for iceberg.
31-
/// TODO: add more and sort them based on some rules.
3231
enum class ErrorKind {
33-
kNoSuchNamespace,
3432
kAlreadyExists,
35-
kNoSuchTable,
3633
kCommitStateUnknown,
37-
kInvalidSchema,
3834
kInvalidArgument,
39-
kIOError,
40-
kNotImplemented,
41-
kUnknownError,
42-
kNotSupported,
4335
kInvalidExpression,
36+
kInvalidSchema,
37+
kIOError,
4438
kJsonParseError,
39+
kNoSuchNamespace,
40+
kNoSuchTable,
4541
kNotFound,
42+
kNotImplemented,
43+
kNotSupported,
44+
kUnknownError,
4645
};
4746

4847
/// \brief Error with a kind and a message.
@@ -63,28 +62,29 @@ using Result = expected<T, E>;
6362

6463
using Status = Result<void>;
6564

66-
/// \brief Create an unexpected error with kNotImplemented
67-
template <typename... Args>
68-
auto NotImplementedError(const std::format_string<Args...> fmt, Args&&... args)
69-
-> unexpected<Error> {
70-
return unexpected<Error>({.kind = ErrorKind::kNotImplemented,
71-
.message = std::format(fmt, std::forward<Args>(args)...)});
72-
}
65+
/// \brief Macro to define error creation functions
66+
#define DEFINE_ERROR_FUNCTION(name) \
67+
template <typename... Args> \
68+
inline auto name(const std::format_string<Args...> fmt, Args&&... args) \
69+
-> unexpected<Error> { \
70+
return unexpected<Error>( \
71+
{ErrorKind::k##name, std::format(fmt, std::forward<Args>(args)...)}); \
72+
}
7373

74-
/// \brief Create an unexpected error with kJsonParseError
75-
template <typename... Args>
76-
auto JsonParseError(const std::format_string<Args...> fmt, Args&&... args)
77-
-> unexpected<Error> {
78-
return unexpected<Error>({.kind = ErrorKind::kJsonParseError,
79-
.message = std::format(fmt, std::forward<Args>(args)...)});
80-
}
74+
DEFINE_ERROR_FUNCTION(AlreadyExists)
75+
DEFINE_ERROR_FUNCTION(CommitStateUnknown)
76+
DEFINE_ERROR_FUNCTION(InvalidArgument)
77+
DEFINE_ERROR_FUNCTION(InvalidExpression)
78+
DEFINE_ERROR_FUNCTION(InvalidSchema)
79+
DEFINE_ERROR_FUNCTION(IOError)
80+
DEFINE_ERROR_FUNCTION(JsonParseError)
81+
DEFINE_ERROR_FUNCTION(NoSuchNamespace)
82+
DEFINE_ERROR_FUNCTION(NoSuchTable)
83+
DEFINE_ERROR_FUNCTION(NotFound)
84+
DEFINE_ERROR_FUNCTION(NotImplemented)
85+
DEFINE_ERROR_FUNCTION(NotSupported)
86+
DEFINE_ERROR_FUNCTION(UnknownError)
8187

82-
/// \brief Create an unexpected error with kInvalidExpression
83-
template <typename... Args>
84-
auto InvalidExpressionError(const std::format_string<Args...> fmt, Args&&... args)
85-
-> unexpected<Error> {
86-
return unexpected<Error>({.kind = ErrorKind::kInvalidExpression,
87-
.message = std::format(fmt, std::forward<Args>(args)...)});
88-
}
88+
#undef DEFINE_ERROR_FUNCTION
8989

9090
} // namespace iceberg

‎src/iceberg/schema_internal.cc‎

Lines changed: 28 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,12 @@
2020
#include "iceberg/schema_internal.h"
2121

2222
#include <cstring>
23-
#include <format>
2423
#include <optional>
2524
#include <string>
2625

2726
#include "iceberg/schema.h"
2827
#include "iceberg/type.h"
28+
#include "iceberg/util/macros.h"
2929

3030
namespace iceberg {
3131

@@ -170,18 +170,14 @@ ArrowErrorCode ToArrowSchema(const Type& type, bool optional, std::string_view n
170170

171171
Status ToArrowSchema(const Schema& schema, ArrowSchema* out) {
172172
if (out == nullptr) [[unlikely]] {
173-
return unexpected<Error>{{.kind = ErrorKind::kInvalidArgument,
174-
.message = "Output Arrow schema cannot be null"}};
173+
return InvalidArgument("Output Arrow schema cannot be null");
175174
}
176175

177176
if (ArrowErrorCode errorCode = ToArrowSchema(schema, /*optional=*/false, /*name=*/"",
178177
/*field_id=*/std::nullopt, out);
179178
errorCode != NANOARROW_OK) {
180-
return unexpected<Error>{
181-
{.kind = ErrorKind::kInvalidSchema,
182-
.message = std::format(
183-
"Failed to convert Iceberg schema to Arrow schema, error code: {}",
184-
errorCode)}};
179+
return InvalidSchema(
180+
"Failed to convert Iceberg schema to Arrow schema, error code: {}", errorCode);
185181
}
186182

187183
return {};
@@ -208,15 +204,12 @@ int32_t GetFieldId(const ArrowSchema& schema) {
208204
Result<std::shared_ptr<Type>> FromArrowSchema(const ArrowSchema& schema) {
209205
auto to_schema_field =
210206
[](const ArrowSchema& schema) -> Result<std::unique_ptr<SchemaField>> {
211-
auto field_type_result = FromArrowSchema(schema);
212-
if (!field_type_result) {
213-
return unexpected<Error>(field_type_result.error());
214-
}
207+
ICEBERG_ASSIGN_OR_RAISE(auto field_type, FromArrowSchema(schema));
215208

216209
auto field_id = GetFieldId(schema);
217210
bool is_optional = (schema.flags & ARROW_FLAG_NULLABLE) != 0;
218-
return std::make_unique<SchemaField>(
219-
field_id, schema.name, std::move(field_type_result.value()), is_optional);
211+
return std::make_unique<SchemaField>(field_id, schema.name, std::move(field_type),
212+
is_optional);
220213
};
221214

222215
ArrowError arrow_error;
@@ -225,10 +218,8 @@ Result<std::shared_ptr<Type>> FromArrowSchema(const ArrowSchema& schema) {
225218
ArrowSchemaView schema_view;
226219
if (auto error_code = ArrowSchemaViewInit(&schema_view, &schema, &arrow_error);
227220
error_code != NANOARROW_OK) {
228-
return unexpected<Error>{
229-
{.kind = ErrorKind::kInvalidSchema,
230-
.message = std::format("Failed to read Arrow schema, code: {}, message: {}",
231-
error_code, arrow_error.message)}};
221+
return InvalidSchema("Failed to read Arrow schema, code: {}, message: {}", error_code,
222+
arrow_error.message);
232223
}
233224

234225
switch (schema_view.type) {
@@ -237,35 +228,24 @@ Result<std::shared_ptr<Type>> FromArrowSchema(const ArrowSchema& schema) {
237228
fields.reserve(schema.n_children);
238229

239230
for (int i = 0; i < schema.n_children; i++) {
240-
auto field_result = to_schema_field(*schema.children[i]);
241-
if (!field_result) {
242-
return unexpected<Error>(field_result.error());
243-
}
244-
fields.emplace_back(std::move(*field_result.value()));
231+
ICEBERG_ASSIGN_OR_RAISE(auto field, to_schema_field(*schema.children[i]));
232+
fields.emplace_back(std::move(*field));
245233
}
246234

247235
return std::make_shared<StructType>(std::move(fields));
248236
}
249237
case NANOARROW_TYPE_LIST: {
250-
auto element_field_result = to_schema_field(*schema.children[0]);
251-
if (!element_field_result) {
252-
return unexpected<Error>(element_field_result.error());
253-
}
254-
return std::make_shared<ListType>(std::move(*element_field_result.value()));
238+
ICEBERG_ASSIGN_OR_RAISE(auto element_field_result,
239+
to_schema_field(*schema.children[0]));
240+
return std::make_shared<ListType>(std::move(*element_field_result));
255241
}
256242
case NANOARROW_TYPE_MAP: {
257-
auto key_field_result = to_schema_field(*schema.children[0]->children[0]);
258-
if (!key_field_result) {
259-
return unexpected<Error>(key_field_result.error());
260-
}
243+
ICEBERG_ASSIGN_OR_RAISE(auto key_field,
244+
to_schema_field(*schema.children[0]->children[0]));
245+
ICEBERG_ASSIGN_OR_RAISE(auto value_field,
246+
to_schema_field(*schema.children[0]->children[1]));
261247

262-
auto value_field_result = to_schema_field(*schema.children[0]->children[1]);
263-
if (!value_field_result) {
264-
return unexpected<Error>(value_field_result.error());
265-
}
266-
267-
return std::make_shared<MapType>(std::move(*key_field_result.value()),
268-
std::move(*value_field_result.value()));
248+
return std::make_shared<MapType>(std::move(*key_field), std::move(*value_field));
269249
}
270250
case NANOARROW_TYPE_BOOL:
271251
return std::make_shared<BooleanType>();
@@ -284,20 +264,16 @@ Result<std::shared_ptr<Type>> FromArrowSchema(const ArrowSchema& schema) {
284264
return std::make_shared<DateType>();
285265
case NANOARROW_TYPE_TIME64:
286266
if (schema_view.time_unit != NANOARROW_TIME_UNIT_MICRO) {
287-
return unexpected<Error>{
288-
{.kind = ErrorKind::kInvalidSchema,
289-
.message = std::format("Unsupported time unit for Arrow time type: {}",
290-
static_cast<int>(schema_view.time_unit))}};
267+
return InvalidSchema("Unsupported time unit for Arrow time type: {}",
268+
static_cast<int>(schema_view.time_unit));
291269
}
292270
return std::make_shared<TimeType>();
293271
case NANOARROW_TYPE_TIMESTAMP: {
294272
bool with_timezone =
295273
schema_view.timezone != nullptr && std::strlen(schema_view.timezone) > 0;
296274
if (schema_view.time_unit != NANOARROW_TIME_UNIT_MICRO) {
297-
return unexpected<Error>{
298-
{.kind = ErrorKind::kInvalidSchema,
299-
.message = std::format("Unsupported time unit for Arrow timestamp type: {}",
300-
static_cast<int>(schema_view.time_unit))}};
275+
return InvalidSchema("Unsupported time unit for Arrow timestamp type: {}",
276+
static_cast<int>(schema_view.time_unit));
301277
}
302278
if (with_timezone) {
303279
return std::make_shared<TimestampTzType>();
@@ -314,18 +290,15 @@ Result<std::shared_ptr<Type>> FromArrowSchema(const ArrowSchema& schema) {
314290
schema_view.extension_name.size_bytes);
315291
extension_name == kArrowUuidExtensionName) {
316292
if (schema_view.fixed_size != 16) {
317-
return unexpected<Error>{{.kind = ErrorKind::kInvalidSchema,
318-
.message = "UUID type must have a fixed size of 16"}};
293+
return InvalidSchema("UUID type must have a fixed size of 16");
319294
}
320295
return std::make_shared<UuidType>();
321296
}
322297
return std::make_shared<FixedType>(schema_view.fixed_size);
323298
}
324299
default:
325-
return unexpected<Error>{
326-
{.kind = ErrorKind::kInvalidSchema,
327-
.message = std::format("Unsupported Arrow type: {}",
328-
ArrowTypeString(schema_view.type))}};
300+
return InvalidSchema("Unsupported Arrow type: {}",
301+
ArrowTypeString(schema_view.type));
329302
}
330303
}
331304

@@ -343,16 +316,10 @@ std::unique_ptr<Schema> FromStructType(StructType&& struct_type,
343316

344317
Result<std::unique_ptr<Schema>> FromArrowSchema(const ArrowSchema& schema,
345318
std::optional<int32_t> schema_id) {
346-
auto type_result = FromArrowSchema(schema);
347-
if (!type_result) {
348-
return unexpected<Error>(type_result.error());
349-
}
319+
ICEBERG_ASSIGN_OR_RAISE(auto type, FromArrowSchema(schema));
350320

351-
auto& type = type_result.value();
352321
if (type->type_id() != TypeId::kStruct) {
353-
return unexpected<Error>{
354-
{.kind = ErrorKind::kInvalidSchema,
355-
.message = "Arrow schema must be a struct type for Iceberg schema"}};
322+
return InvalidSchema("Arrow schema must be a struct type for Iceberg schema");
356323
}
357324

358325
auto& struct_type = static_cast<StructType&>(*type);

0 commit comments

Comments
 (0)