| /* | |
| * 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. | |
| */ | |
| #include <algorithm> | |
| #include <concepts> | |
| #include <cstdint> | |
| #include <format> | |
| #include <map> | |
| #include <optional> | |
| #include <regex> | |
| #include <string> | |
| #include <unordered_set> | |
| #include <utility> | |
| #include <nlohmann/json.hpp> | |
| #include "iceberg/constants.h" | |
| #include "iceberg/expression/expressions.h" | |
| #include "iceberg/expression/json_serde_internal.h" | |
| #include "iceberg/expression/literal.h" | |
| #include "iceberg/file_format.h" | |
| #include "iceberg/json_serde_internal.h" | |
| #include "iceberg/manifest/manifest_entry.h" | |
| #include "iceberg/name_mapping.h" | |
| #include "iceberg/partition_field.h" | |
| #include "iceberg/partition_spec.h" | |
| #include "iceberg/result.h" | |
| #include "iceberg/row/partition_values.h" | |
| #include "iceberg/schema.h" | |
| #include "iceberg/schema_field.h" | |
| #include "iceberg/snapshot.h" | |
| #include "iceberg/sort_order.h" | |
| #include "iceberg/statistics_file.h" | |
| #include "iceberg/table_identifier.h" | |
| #include "iceberg/table_metadata.h" | |
| #include "iceberg/table_properties.h" | |
| #include "iceberg/table_requirement.h" | |
| #include "iceberg/table_scan.h" | |
| #include "iceberg/table_update.h" | |
| #include "iceberg/transform.h" | |
| #include "iceberg/type.h" | |
| #include "iceberg/util/base64.h" | |
| #include "iceberg/util/checked_cast.h" | |
| #include "iceberg/util/formatter.h" // IWYU pragma: keep | |
| #include "iceberg/util/json_util_internal.h" | |
| #include "iceberg/util/macros.h" | |
| #include "iceberg/util/string_util.h" | |
| #include "iceberg/util/temporal_util.h" | |
| #include "iceberg/util/timepoint.h" | |
| namespace iceberg { | |
| namespace { | |
| // Transform constants | |
| constexpr std::string_view kTransform = "transform"; | |
| constexpr std::string_view kSourceId = "source-id"; | |
| constexpr std::string_view kDirection = "direction"; | |
| constexpr std::string_view kNullOrder = "null-order"; | |
| // Sort order constants | |
| constexpr std::string_view kOrderId = "order-id"; | |
| constexpr std::string_view kFields = "fields"; | |
| // Schema constants | |
| constexpr std::string_view kSchemaId = "schema-id"; | |
| constexpr std::string_view kIdentifierFieldIds = "identifier-field-ids"; | |
| // Type constants | |
| constexpr std::string_view kType = "type"; | |
| constexpr std::string_view kStruct = "struct"; | |
| constexpr std::string_view kList = "list"; | |
| constexpr std::string_view kMap = "map"; | |
| constexpr std::string_view kElement = "element"; | |
| constexpr std::string_view kKey = "key"; | |
| constexpr std::string_view kValue = "value"; | |
| constexpr std::string_view kDoc = "doc"; | |
| constexpr std::string_view kName = "name"; | |
| constexpr std::string_view kNamespace = "namespace"; | |
| constexpr std::string_view kNames = "names"; | |
| constexpr std::string_view kId = "id"; | |
| constexpr std::string_view kInitialDefault = "initial-default"; | |
| constexpr std::string_view kWriteDefault = "write-default"; | |
| constexpr std::string_view kFieldId = "field-id"; | |
| constexpr std::string_view kElementId = "element-id"; | |
| constexpr std::string_view kKeyId = "key-id"; | |
| constexpr std::string_view kValueId = "value-id"; | |
| constexpr std::string_view kRequired = "required"; | |
| constexpr std::string_view kElementRequired = "element-required"; | |
| constexpr std::string_view kValueRequired = "value-required"; | |
| constexpr std::string_view kEncryptedKeyMetadata = "encrypted-key-metadata"; | |
| constexpr std::string_view kEncryptedById = "encrypted-by-id"; | |
| // Snapshot constants | |
| constexpr std::string_view kSpecId = "spec-id"; | |
| constexpr std::string_view kSnapshotId = "snapshot-id"; | |
| constexpr std::string_view kParentSnapshotId = "parent-snapshot-id"; | |
| constexpr std::string_view kSequenceNumber = "sequence-number"; | |
| constexpr std::string_view kTimestampMs = "timestamp-ms"; | |
| constexpr std::string_view kManifestList = "manifest-list"; | |
| constexpr std::string_view kSummary = "summary"; | |
| constexpr std::string_view kFirstRowId = "first-row-id"; | |
| constexpr std::string_view kAddedRows = "added-rows"; | |
| constexpr std::string_view kMinSnapshotsToKeep = "min-snapshots-to-keep"; | |
| constexpr std::string_view kMaxSnapshotAgeMs = "max-snapshot-age-ms"; | |
| constexpr std::string_view kMaxRefAgeMs = "max-ref-age-ms"; | |
| const std::unordered_set<std::string_view> kValidSnapshotSummaryFields = { | |
| SnapshotSummaryFields::kOperation, | |
| SnapshotSummaryFields::kAddedDataFiles, | |
| SnapshotSummaryFields::kDeletedDataFiles, | |
| SnapshotSummaryFields::kTotalDataFiles, | |
| SnapshotSummaryFields::kAddedDeleteFiles, | |
| SnapshotSummaryFields::kAddedEqDeleteFiles, | |
| SnapshotSummaryFields::kRemovedEqDeleteFiles, | |
| SnapshotSummaryFields::kAddedPosDeleteFiles, | |
| SnapshotSummaryFields::kRemovedPosDeleteFiles, | |
| SnapshotSummaryFields::kAddedDVs, | |
| SnapshotSummaryFields::kRemovedDVs, | |
| SnapshotSummaryFields::kRemovedDeleteFiles, | |
| SnapshotSummaryFields::kTotalDeleteFiles, | |
| SnapshotSummaryFields::kAddedRecords, | |
| SnapshotSummaryFields::kDeletedRecords, | |
| SnapshotSummaryFields::kTotalRecords, | |
| SnapshotSummaryFields::kAddedFileSize, | |
| SnapshotSummaryFields::kRemovedFileSize, | |
| SnapshotSummaryFields::kTotalFileSize, | |
| SnapshotSummaryFields::kAddedPosDeletes, | |
| SnapshotSummaryFields::kRemovedPosDeletes, | |
| SnapshotSummaryFields::kTotalPosDeletes, | |
| SnapshotSummaryFields::kAddedEqDeletes, | |
| SnapshotSummaryFields::kRemovedEqDeletes, | |
| SnapshotSummaryFields::kTotalEqDeletes, | |
| SnapshotSummaryFields::kDeletedDuplicatedFiles, | |
| SnapshotSummaryFields::kChangedPartitionCountProp, | |
| SnapshotSummaryFields::kWAPId, | |
| SnapshotSummaryFields::kPublishedWAPId, | |
| SnapshotSummaryFields::kSourceSnapshotId, | |
| SnapshotSummaryFields::kEngineName, | |
| SnapshotSummaryFields::kEngineVersion}; | |
| const std::unordered_set<std::string_view> kValidDataOperation = { | |
| DataOperation::kAppend, DataOperation::kReplace, DataOperation::kOverwrite, | |
| DataOperation::kDelete}; | |
| // TableMetadata constants | |
| constexpr std::string_view kFormatVersion = "format-version"; | |
| constexpr std::string_view kTableUuid = "table-uuid"; | |
| constexpr std::string_view kLocation = "location"; | |
| constexpr std::string_view kLastSequenceNumber = "last-sequence-number"; | |
| constexpr std::string_view kLastUpdatedMs = "last-updated-ms"; | |
| constexpr std::string_view kLastColumnId = "last-column-id"; | |
| constexpr std::string_view kSchema = "schema"; | |
| constexpr std::string_view kSchemas = "schemas"; | |
| constexpr std::string_view kCurrentSchemaId = "current-schema-id"; | |
| constexpr std::string_view kPartitionSpec = "partition-spec"; | |
| constexpr std::string_view kPartitionSpecs = "partition-specs"; | |
| constexpr std::string_view kDefaultSpecId = "default-spec-id"; | |
| constexpr std::string_view kLastPartitionId = "last-partition-id"; | |
| constexpr std::string_view kProperties = "properties"; | |
| constexpr std::string_view kCurrentSnapshotId = "current-snapshot-id"; | |
| constexpr std::string_view kSnapshots = "snapshots"; | |
| constexpr std::string_view kSnapshotLog = "snapshot-log"; | |
| constexpr std::string_view kMetadataLog = "metadata-log"; | |
| constexpr std::string_view kSortOrders = "sort-orders"; | |
| constexpr std::string_view kDefaultSortOrderId = "default-sort-order-id"; | |
| constexpr std::string_view kRefs = "refs"; | |
| constexpr std::string_view kStatistics = "statistics"; | |
| constexpr std::string_view kPartitionStatistics = "partition-statistics"; | |
| constexpr std::string_view kNextRowId = "next-row-id"; | |
| constexpr std::string_view kEncryptionKeys = "encryption-keys"; | |
| constexpr std::string_view kMetadataFile = "metadata-file"; | |
| constexpr std::string_view kStatisticsPath = "statistics-path"; | |
| constexpr std::string_view kFileSizeInBytes = "file-size-in-bytes"; | |
| constexpr std::string_view kFileFooterSizeInBytes = "file-footer-size-in-bytes"; | |
| constexpr std::string_view kBlobMetadata = "blob-metadata"; | |
| // TableUpdate action constants | |
| constexpr std::string_view kAction = "action"; | |
| constexpr std::string_view kActionAssignUUID = "assign-uuid"; | |
| constexpr std::string_view kActionUpgradeFormatVersion = "upgrade-format-version"; | |
| constexpr std::string_view kActionAddSchema = "add-schema"; | |
| constexpr std::string_view kActionSetCurrentSchema = "set-current-schema"; | |
| constexpr std::string_view kActionAddPartitionSpec = "add-spec"; | |
| constexpr std::string_view kActionSetDefaultPartitionSpec = "set-default-spec"; | |
| constexpr std::string_view kActionRemovePartitionSpecs = "remove-partition-specs"; | |
| constexpr std::string_view kActionRemoveSchemas = "remove-schemas"; | |
| constexpr std::string_view kActionAddSortOrder = "add-sort-order"; | |
| constexpr std::string_view kActionSetDefaultSortOrder = "set-default-sort-order"; | |
| constexpr std::string_view kActionAddSnapshot = "add-snapshot"; | |
| constexpr std::string_view kActionRemoveSnapshots = "remove-snapshots"; | |
| constexpr std::string_view kActionRemoveSnapshotRef = "remove-snapshot-ref"; | |
| constexpr std::string_view kActionSetSnapshotRef = "set-snapshot-ref"; | |
| constexpr std::string_view kActionSetProperties = "set-properties"; | |
| constexpr std::string_view kActionRemoveProperties = "remove-properties"; | |
| constexpr std::string_view kActionSetLocation = "set-location"; | |
| constexpr std::string_view kActionSetStatistics = "set-statistics"; | |
| constexpr std::string_view kActionRemoveStatistics = "remove-statistics"; | |
| constexpr std::string_view kActionSetPartitionStatistics = "set-partition-statistics"; | |
| constexpr std::string_view kActionRemovePartitionStatistics = | |
| "remove-partition-statistics"; | |
| constexpr std::string_view kActionAddEncryptionKey = "add-encryption-key"; | |
| constexpr std::string_view kActionRemoveEncryptionKey = "remove-encryption-key"; | |
| // TableUpdate field constants | |
| constexpr std::string_view kUUID = "uuid"; | |
| constexpr std::string_view kSpec = "spec"; | |
| constexpr std::string_view kSpecIds = "spec-ids"; | |
| constexpr std::string_view kSchemaIds = "schema-ids"; | |
| constexpr std::string_view kSortOrder = "sort-order"; | |
| constexpr std::string_view kSortOrderId = "sort-order-id"; | |
| constexpr std::string_view kSnapshot = "snapshot"; | |
| constexpr std::string_view kSnapshotIds = "snapshot-ids"; | |
| constexpr std::string_view kRefName = "ref-name"; | |
| constexpr std::string_view kRef = "ref"; | |
| constexpr std::string_view kUpdates = "updates"; | |
| constexpr std::string_view kRemovals = "removals"; | |
| constexpr std::string_view kUpdated = "updated"; | |
| constexpr std::string_view kRemoved = "removed"; | |
| constexpr std::string_view kEncryptionKey = "encryption-key"; | |
| // TableRequirement type constants | |
| constexpr std::string_view kRequirementAssertDoesNotExist = "assert-create"; | |
| constexpr std::string_view kRequirementAssertUUID = "assert-table-uuid"; | |
| constexpr std::string_view kRequirementAssertRefSnapshotID = "assert-ref-snapshot-id"; | |
| constexpr std::string_view kRequirementAssertLastAssignedFieldId = | |
| "assert-last-assigned-field-id"; | |
| constexpr std::string_view kRequirementAssertCurrentSchemaID = "assert-current-schema-id"; | |
| constexpr std::string_view kRequirementAssertLastAssignedPartitionId = | |
| "assert-last-assigned-partition-id"; | |
| constexpr std::string_view kRequirementAssertDefaultSpecID = "assert-default-spec-id"; | |
| constexpr std::string_view kRequirementAssertDefaultSortOrderID = | |
| "assert-default-sort-order-id"; | |
| constexpr std::string_view kLastAssignedFieldId = "last-assigned-field-id"; | |
| constexpr std::string_view kLastAssignedPartitionId = "last-assigned-partition-id"; | |
| // DataFile / FileScanTask JSON (Iceberg REST ContentFile field names) | |
| constexpr std::string_view kContent = "content"; | |
| constexpr std::string_view kContentData = "data"; | |
| constexpr std::string_view kContentPositionDeletes = "position-deletes"; | |
| constexpr std::string_view kContentEqualityDeletes = "equality-deletes"; | |
| constexpr std::string_view kFilePath = "file-path"; | |
| constexpr std::string_view kFileFormat = "file-format"; | |
| constexpr std::string_view kPartition = "partition"; | |
| constexpr std::string_view kRecordCount = "record-count"; | |
| constexpr std::string_view kColumnSizes = "column-sizes"; | |
| constexpr std::string_view kValueCounts = "value-counts"; | |
| constexpr std::string_view kNullValueCounts = "null-value-counts"; | |
| constexpr std::string_view kNanValueCounts = "nan-value-counts"; | |
| constexpr std::string_view kLowerBounds = "lower-bounds"; | |
| constexpr std::string_view kUpperBounds = "upper-bounds"; | |
| constexpr std::string_view kKeyMetadata = "key-metadata"; | |
| constexpr std::string_view kSplitOffsets = "split-offsets"; | |
| constexpr std::string_view kEqualityIds = "equality-ids"; | |
| constexpr std::string_view kReferencedDataFile = "referenced-data-file"; | |
| constexpr std::string_view kContentOffset = "content-offset"; | |
| constexpr std::string_view kContentSizeInBytes = "content-size-in-bytes"; | |
| constexpr std::string_view kDataFile = "data-file"; | |
| constexpr std::string_view kDeleteFiles = "delete-files"; | |
| constexpr std::string_view kDeleteFileReferences = "delete-file-references"; | |
| constexpr std::string_view kResidualFilter = "residual-filter"; | |
| constexpr std::string_view kTaskType = "task-type"; | |
| constexpr std::string_view kFileScanTaskType = "file-scan-task"; | |
| constexpr std::string_view kStart = "start"; | |
| constexpr std::string_view kLength = "length"; | |
| constexpr std::string_view kMapKeys = "keys"; | |
| constexpr std::string_view kMapValues = "values"; | |
| template <std::signed_integral Value> | |
| Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json, | |
| std::string_view key) { | |
| std::map<int32_t, Value> result; | |
| if (!json.contains(key) || json.at(key).is_null()) { | |
| return result; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto map_json, GetJsonValue<nlohmann::json>(json, key)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto keys_json, | |
| GetJsonValue<nlohmann::json>(map_json, kMapKeys)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto values_json, | |
| GetJsonValue<nlohmann::json>(map_json, kMapValues)); | |
| if (!keys_json.is_array() || !values_json.is_array()) { | |
| return JsonParseError("'{}' map keys and values must be arrays", key); | |
| } | |
| if (keys_json.size() != values_json.size()) { | |
| return JsonParseError("'{}' map keys and values have different lengths", key); | |
| } | |
| for (size_t i = 0; i < keys_json.size(); ++i) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto field_id, | |
| IntegerFromJson<int32_t>(keys_json[i], std::format("'{}' map key", key))); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto value, | |
| IntegerFromJson<Value>(values_json[i], std::format("'{}' map value", key))); | |
| if (!result.emplace(field_id, value).second) { | |
| return JsonParseError("'{}' map contains duplicate key {}", key, field_id); | |
| } | |
| } | |
| return result; | |
| } | |
| template <typename Value> | |
| void SetKeyValueMap(nlohmann::json& json, std::string_view key, | |
| const std::map<int32_t, Value>& map) { | |
| if (map.empty()) { | |
| return; | |
| } | |
| std::vector<int32_t> keys; | |
| std::vector<Value> values; | |
| keys.reserve(map.size()); | |
| values.reserve(map.size()); | |
| for (const auto& [field_id, value] : map) { | |
| keys.push_back(field_id); | |
| values.push_back(value); | |
| } | |
| json[key] = {{kMapKeys, std::move(keys)}, {kMapValues, std::move(values)}}; | |
| } | |
| std::string BytesToHex(const std::vector<uint8_t>& bytes) { | |
| std::string hex; | |
| hex.reserve(bytes.size() * 2); | |
| for (uint8_t byte : bytes) { | |
| hex += std::format("{:02X}", byte); | |
| } | |
| return hex; | |
| } | |
| Result<std::map<int32_t, std::vector<uint8_t>>> BytesMapFromJson( | |
| const nlohmann::json& json, std::string_view key) { | |
| std::map<int32_t, std::vector<uint8_t>> result; | |
| if (!json.contains(key) || json.at(key).is_null()) { | |
| return result; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto map_json, GetJsonValue<nlohmann::json>(json, key)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto keys_json, | |
| GetJsonValue<nlohmann::json>(map_json, kMapKeys)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto values_json, | |
| GetJsonValue<nlohmann::json>(map_json, kMapValues)); | |
| if (!keys_json.is_array() || !values_json.is_array()) { | |
| return JsonParseError("'{}' map keys and values must be arrays", key); | |
| } | |
| if (keys_json.size() != values_json.size()) { | |
| return JsonParseError("'{}' map keys and values have different lengths", key); | |
| } | |
| for (size_t i = 0; i < keys_json.size(); ++i) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto field_id, | |
| IntegerFromJson<int32_t>(keys_json[i], std::format("'{}' map key", key))); | |
| if (!values_json[i].is_string()) { | |
| return JsonParseError("'{}' map value must be a string, but is {}", key, | |
| values_json[i].type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto bytes, StringUtils::HexStringToBytes(values_json[i].get<std::string>())); | |
| if (!result.emplace(field_id, std::move(bytes)).second) { | |
| return JsonParseError("'{}' map contains duplicate key {}", key, field_id); | |
| } | |
| } | |
| return result; | |
| } | |
| void SetBytesMap(nlohmann::json& json, std::string_view key, | |
| const std::map<int32_t, std::vector<uint8_t>>& map) { | |
| if (map.empty()) { | |
| return; | |
| } | |
| std::vector<int32_t> keys; | |
| std::vector<std::string> values; | |
| keys.reserve(map.size()); | |
| values.reserve(map.size()); | |
| for (const auto& [field_id, value] : map) { | |
| keys.push_back(field_id); | |
| values.push_back(BytesToHex(value)); | |
| } | |
| json[key] = {{kMapKeys, std::move(keys)}, {kMapValues, std::move(values)}}; | |
| } | |
| Result<Literal> PartitionLiteralFromJson(const nlohmann::json* value, | |
| const SchemaField& field) { | |
| if (!field.type() || !field.type()->is_primitive()) { | |
| return InvalidSchema("Partition field {} must have a primitive type", | |
| field.field_id()); | |
| } | |
| auto primitive_type = internal::checked_pointer_cast<PrimitiveType>(field.type()); | |
| if (value == nullptr || value->is_null()) { | |
| return Literal::Null(std::move(primitive_type)); | |
| } | |
| return LiteralFromJson(*value, primitive_type.get()); | |
| } | |
| Result<PartitionValues> PartitionValuesFromJson(const nlohmann::json& json, | |
| const StructType& partition_type) { | |
| const auto& fields = partition_type.fields(); | |
| std::vector<Literal> values; | |
| values.reserve(fields.size()); | |
| if (json.is_array()) { | |
| if (json.size() != fields.size()) { | |
| return JsonParseError("Invalid partition data size: expected = {}, actual = {}", | |
| fields.size(), json.size()); | |
| } | |
| for (size_t pos = 0; pos < fields.size(); ++pos) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto value, | |
| PartitionLiteralFromJson(&json[pos], fields[pos])); | |
| values.push_back(std::move(value)); | |
| } | |
| } else if (json.is_object()) { | |
| if (json.size() > fields.size()) { | |
| return JsonParseError("Invalid partition data size: expected <= {}, actual = {}", | |
| fields.size(), json.size()); | |
| } | |
| for (const auto& field : fields) { | |
| const auto it = json.find(std::to_string(field.field_id())); | |
| const nlohmann::json* value = it == json.end() ? nullptr : &it.value(); | |
| ICEBERG_ASSIGN_OR_RAISE(auto literal, PartitionLiteralFromJson(value, field)); | |
| values.push_back(std::move(literal)); | |
| } | |
| } else { | |
| return JsonParseError( | |
| "Invalid partition data for content file: expected array or object (JSON value " | |
| "of type {})", | |
| json.type_name()); | |
| } | |
| return PartitionValues(std::move(values)); | |
| } | |
| Result<nlohmann::json> DataFileToJsonUnchecked(const DataFile& data_file) { | |
| nlohmann::json json; | |
| switch (data_file.content) { | |
| case DataFile::Content::kData: | |
| json[kContent] = kContentData; | |
| break; | |
| case DataFile::Content::kPositionDeletes: | |
| json[kContent] = kContentPositionDeletes; | |
| break; | |
| case DataFile::Content::kEqualityDeletes: | |
| json[kContent] = kContentEqualityDeletes; | |
| break; | |
| } | |
| json[kFilePath] = data_file.file_path; | |
| json[kFileFormat] = ToString(data_file.file_format); | |
| if (!data_file.partition_spec_id.has_value()) { | |
| return ValidationFailed("Cannot serialize content file without 'spec-id'"); | |
| } | |
| json[kSpecId] = data_file.partition_spec_id.value(); | |
| nlohmann::json partition_json = nlohmann::json::array(); | |
| for (const auto& literal : data_file.partition.values()) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto lit_json, ToJson(literal)); | |
| partition_json.push_back(std::move(lit_json)); | |
| } | |
| json[kPartition] = std::move(partition_json); | |
| json[kRecordCount] = data_file.record_count; | |
| json[kFileSizeInBytes] = data_file.file_size_in_bytes; | |
| SetKeyValueMap(json, kColumnSizes, data_file.column_sizes); | |
| SetKeyValueMap(json, kValueCounts, data_file.value_counts); | |
| SetKeyValueMap(json, kNullValueCounts, data_file.null_value_counts); | |
| SetKeyValueMap(json, kNanValueCounts, data_file.nan_value_counts); | |
| SetBytesMap(json, kLowerBounds, data_file.lower_bounds); | |
| SetBytesMap(json, kUpperBounds, data_file.upper_bounds); | |
| if (!data_file.key_metadata.empty()) { | |
| json[kKeyMetadata] = BytesToHex(data_file.key_metadata); | |
| } | |
| if (!data_file.split_offsets.empty()) { | |
| json[kSplitOffsets] = data_file.split_offsets; | |
| } | |
| if (!data_file.equality_ids.empty()) { | |
| json[kEqualityIds] = data_file.equality_ids; | |
| } | |
| if (data_file.sort_order_id.has_value()) { | |
| json[kSortOrderId] = data_file.sort_order_id.value(); | |
| } | |
| if (data_file.first_row_id.has_value()) { | |
| json[kFirstRowId] = data_file.first_row_id.value(); | |
| } | |
| if (data_file.referenced_data_file.has_value()) { | |
| json[kReferencedDataFile] = data_file.referenced_data_file.value(); | |
| } | |
| if (data_file.content_offset.has_value()) { | |
| json[kContentOffset] = data_file.content_offset.value(); | |
| } | |
| if (data_file.content_size_in_bytes.has_value()) { | |
| json[kContentSizeInBytes] = data_file.content_size_in_bytes.value(); | |
| } | |
| return json; | |
| } | |
| } // namespace | |
| nlohmann::json ToJson(const SortField& sort_field) { | |
| nlohmann::json json; | |
| json[kTransform] = std::format("{}", *sort_field.transform()); | |
| json[kSourceId] = sort_field.source_id(); | |
| json[kDirection] = std::format("{}", sort_field.direction()); | |
| json[kNullOrder] = std::format("{}", sort_field.null_order()); | |
| return json; | |
| } | |
| nlohmann::json ToJson(const SortOrder& sort_order) { | |
| nlohmann::json json; | |
| json[kOrderId] = sort_order.order_id(); | |
| nlohmann::json fields_json = nlohmann::json::array(); | |
| for (const auto& field : sort_order.fields()) { | |
| fields_json.push_back(ToJson(field)); | |
| } | |
| json[kFields] = fields_json; | |
| return json; | |
| } | |
| Result<std::unique_ptr<SortField>> SortFieldFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto source_id, GetJsonValue<int32_t>(json, kSourceId)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto transform, | |
| GetJsonValue<std::string>(json, kTransform).and_then(TransformFromString)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto direction, | |
| GetJsonValue<std::string>(json, kDirection).and_then(SortDirectionFromString)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto null_order, | |
| GetJsonValue<std::string>(json, kNullOrder).and_then(NullOrderFromString)); | |
| return std::make_unique<SortField>(source_id, std::move(transform), direction, | |
| null_order); | |
| } | |
| namespace { | |
| struct ParsedSortOrder { | |
| int32_t order_id; | |
| std::vector<SortField> fields; | |
| }; | |
| Result<ParsedSortOrder> ParseSortOrder(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto order_id, GetJsonValue<int32_t>(json, kOrderId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto fields, GetJsonValue<nlohmann::json>(json, kFields)); | |
| std::vector<SortField> sort_fields; | |
| for (const auto& field_json : fields) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_field, SortFieldFromJson(field_json)); | |
| sort_fields.push_back(std::move(*sort_field)); | |
| } | |
| return ParsedSortOrder{.order_id = order_id, .fields = std::move(sort_fields)}; | |
| } | |
| } // namespace | |
| Result<std::unique_ptr<SortOrder>> SortOrderFromJson( | |
| const nlohmann::json& json, const std::shared_ptr<Schema>& current_schema, | |
| int32_t default_sort_order_id) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto parsed, ParseSortOrder(json)); | |
| if (parsed.order_id == default_sort_order_id) { | |
| return SortOrder::Make(*current_schema, parsed.order_id, std::move(parsed.fields)); | |
| } | |
| return SortOrder::Make(parsed.order_id, std::move(parsed.fields)); | |
| } | |
| Result<std::unique_ptr<SortOrder>> SortOrderFromJson( | |
| const nlohmann::json& json, const std::shared_ptr<Schema>& current_schema) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto parsed, ParseSortOrder(json)); | |
| return SortOrder::Make(*current_schema, parsed.order_id, std::move(parsed.fields)); | |
| } | |
| Result<std::unique_ptr<SortOrder>> SortOrderFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto parsed, ParseSortOrder(json)); | |
| return SortOrder::Make(parsed.order_id, std::move(parsed.fields)); | |
| } | |
| Result<nlohmann::json> ToJson(const SchemaField& field) { | |
| nlohmann::json json; | |
| json[kId] = field.field_id(); | |
| json[kName] = field.name(); | |
| json[kRequired] = !field.optional(); | |
| ICEBERG_ASSIGN_OR_RAISE(json[kType], ToJson(*field.type())); | |
| if (!field.doc().empty()) { | |
| json[kDoc] = field.doc(); | |
| } | |
| if (field.initial_default() != nullptr) { | |
| ICEBERG_ASSIGN_OR_RAISE(json[kInitialDefault], ToJson(*field.initial_default())); | |
| } | |
| if (field.write_default() != nullptr) { | |
| ICEBERG_ASSIGN_OR_RAISE(json[kWriteDefault], ToJson(*field.write_default())); | |
| } | |
| return json; | |
| } | |
| Result<nlohmann::json> ToJson(const Type& type) { | |
| switch (type.type_id()) { | |
| case TypeId::kStruct: { | |
| const auto& struct_type = internal::checked_cast<const StructType&>(type); | |
| nlohmann::json json; | |
| json[kType] = kStruct; | |
| nlohmann::json fields_json = nlohmann::json::array(); | |
| for (const auto& field : struct_type.fields()) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto field_json, ToJson(field)); | |
| fields_json.push_back(std::move(field_json)); | |
| } | |
| json[kFields] = fields_json; | |
| return json; | |
| } | |
| case TypeId::kList: { | |
| const auto& list_type = internal::checked_cast<const ListType&>(type); | |
| nlohmann::json json; | |
| json[kType] = kList; | |
| const auto& element_field = list_type.fields().front(); | |
| json[kElementId] = element_field.field_id(); | |
| json[kElementRequired] = !element_field.optional(); | |
| ICEBERG_ASSIGN_OR_RAISE(json[kElement], ToJson(*element_field.type())); | |
| return json; | |
| } | |
| case TypeId::kMap: { | |
| const auto& map_type = internal::checked_cast<const MapType&>(type); | |
| nlohmann::json json; | |
| json[std::string(kType)] = kMap; | |
| const auto& key_field = map_type.key(); | |
| json[kKeyId] = key_field.field_id(); | |
| ICEBERG_ASSIGN_OR_RAISE(json[kKey], ToJson(*key_field.type())); | |
| const auto& value_field = map_type.value(); | |
| json[kValueId] = value_field.field_id(); | |
| json[kValueRequired] = !value_field.optional(); | |
| ICEBERG_ASSIGN_OR_RAISE(json[kValue], ToJson(*value_field.type())); | |
| return json; | |
| } | |
| case TypeId::kBoolean: | |
| return "boolean"; | |
| case TypeId::kInt: | |
| return "int"; | |
| case TypeId::kLong: | |
| return "long"; | |
| case TypeId::kFloat: | |
| return "float"; | |
| case TypeId::kDouble: | |
| return "double"; | |
| case TypeId::kDecimal: { | |
| const auto& decimal_type = internal::checked_cast<const DecimalType&>(type); | |
| return std::format("decimal({},{})", decimal_type.precision(), | |
| decimal_type.scale()); | |
| } | |
| case TypeId::kDate: | |
| return "date"; | |
| case TypeId::kTime: | |
| return "time"; | |
| case TypeId::kTimestamp: | |
| return "timestamp"; | |
| case TypeId::kTimestampTz: | |
| return "timestamptz"; | |
| case TypeId::kTimestampNs: | |
| return "timestamp_ns"; | |
| case TypeId::kTimestampTzNs: | |
| return "timestamptz_ns"; | |
| case TypeId::kString: | |
| return "string"; | |
| case TypeId::kBinary: | |
| return "binary"; | |
| case TypeId::kFixed: { | |
| const auto& fixed_type = internal::checked_cast<const FixedType&>(type); | |
| return std::format("fixed[{}]", fixed_type.length()); | |
| } | |
| case TypeId::kUuid: | |
| return "uuid"; | |
| case TypeId::kUnknown: | |
| return "unknown"; | |
| case TypeId::kVariant: | |
| return "variant"; | |
| case TypeId::kGeometry: | |
| return type.ToString(); | |
| case TypeId::kGeography: | |
| return type.ToString(); | |
| } | |
| std::unreachable(); | |
| } | |
| Result<nlohmann::json> ToJson(const Schema& schema) { | |
| ICEBERG_ASSIGN_OR_RAISE(nlohmann::json json, | |
| ToJson(internal::checked_cast<const Type&>(schema))); | |
| json[kSchemaId] = schema.schema_id(); | |
| if (!schema.IdentifierFieldIds().empty()) { | |
| json[kIdentifierFieldIds] = schema.IdentifierFieldIds(); | |
| } | |
| return json; | |
| } | |
| Result<std::string> ToJsonString(const Schema& schema) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto json, ToJson(schema)); | |
| return ToJsonString(json); | |
| } | |
| nlohmann::json ToJson(const SnapshotRef& ref) { | |
| nlohmann::json json; | |
| json[kSnapshotId] = ref.snapshot_id; | |
| json[kType] = std::format("{}", ref.type()); | |
| if (ref.type() == SnapshotRefType::kBranch) { | |
| const auto& branch = std::get<SnapshotRef::Branch>(ref.retention); | |
| SetOptionalField(json, kMinSnapshotsToKeep, branch.min_snapshots_to_keep); | |
| SetOptionalField(json, kMaxSnapshotAgeMs, branch.max_snapshot_age_ms); | |
| SetOptionalField(json, kMaxRefAgeMs, branch.max_ref_age_ms); | |
| } else if (ref.type() == SnapshotRefType::kTag) { | |
| const auto& tag = std::get<SnapshotRef::Tag>(ref.retention); | |
| SetOptionalField(json, kMaxRefAgeMs, tag.max_ref_age_ms); | |
| } | |
| return json; | |
| } | |
| nlohmann::json ToJson(const Snapshot& snapshot) { | |
| nlohmann::json json; | |
| json[kSnapshotId] = snapshot.snapshot_id; | |
| SetOptionalField(json, kParentSnapshotId, snapshot.parent_snapshot_id); | |
| if (snapshot.sequence_number > TableMetadata::kInitialSequenceNumber) { | |
| json[kSequenceNumber] = snapshot.sequence_number; | |
| } | |
| json[kTimestampMs] = UnixMsFromTimePointMs(snapshot.timestamp_ms); | |
| json[kManifestList] = snapshot.manifest_list; | |
| // If there is an operation, write the summary map | |
| if (snapshot.Operation().has_value()) { | |
| json[kSummary] = snapshot.summary; | |
| } | |
| SetOptionalField(json, kSchemaId, snapshot.schema_id); | |
| SetOptionalField(json, kFirstRowId, snapshot.first_row_id); | |
| if (snapshot.first_row_id.has_value()) { | |
| SetOptionalField(json, kAddedRows, snapshot.added_rows); | |
| } | |
| SetOptionalField(json, kKeyId, snapshot.key_id); | |
| return json; | |
| } | |
| namespace { | |
| Result<std::unique_ptr<Type>> StructTypeFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto json_fields, GetJsonValue<nlohmann::json>(json, kFields)); | |
| std::vector<SchemaField> fields; | |
| for (const auto& field_json : json_fields) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto field, FieldFromJson(field_json)); | |
| fields.emplace_back(std::move(*field)); | |
| } | |
| return std::make_unique<StructType>(std::move(fields)); | |
| } | |
| Result<std::unique_ptr<Type>> ListTypeFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto element_type, | |
| GetJsonValue<nlohmann::json>(json, kElement).and_then(TypeFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto element_id, GetJsonInteger<int32_t>(json, kElementId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto element_required, | |
| GetJsonValue<bool>(json, kElementRequired)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, ListType::Make(SchemaField( | |
| element_id, std::string(ListType::kElementName), | |
| std::move(element_type), !element_required))); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| Result<std::unique_ptr<Type>> MapTypeFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto key_type, GetJsonValue<nlohmann::json>(json, kKey).and_then(TypeFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto value_type, GetJsonValue<nlohmann::json>(json, kValue).and_then(TypeFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonInteger<int32_t>(json, kKeyId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto value_id, GetJsonInteger<int32_t>(json, kValueId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto value_required, GetJsonValue<bool>(json, kValueRequired)); | |
| SchemaField key_field(key_id, std::string(MapType::kKeyName), std::move(key_type), | |
| /*optional=*/false); | |
| SchemaField value_field(value_id, std::string(MapType::kValueName), | |
| std::move(value_type), !value_required); | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, | |
| MapType::Make(std::move(key_field), std::move(value_field))); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| } // namespace | |
| Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json) { | |
| if (json.is_string()) { | |
| const auto type_name = json.get<std::string>(); | |
| const auto normalized_type_name = StringUtils::ToLower(type_name); | |
| if (normalized_type_name == "boolean") { | |
| return std::make_unique<BooleanType>(); | |
| } else if (normalized_type_name == "int") { | |
| return std::make_unique<IntType>(); | |
| } else if (normalized_type_name == "long") { | |
| return std::make_unique<LongType>(); | |
| } else if (normalized_type_name == "float") { | |
| return std::make_unique<FloatType>(); | |
| } else if (normalized_type_name == "double") { | |
| return std::make_unique<DoubleType>(); | |
| } else if (normalized_type_name == "date") { | |
| return std::make_unique<DateType>(); | |
| } else if (normalized_type_name == "time") { | |
| return std::make_unique<TimeType>(); | |
| } else if (normalized_type_name == "timestamp") { | |
| return std::make_unique<TimestampType>(); | |
| } else if (normalized_type_name == "timestamptz") { | |
| return std::make_unique<TimestampTzType>(); | |
| } else if (normalized_type_name == "timestamp_ns") { | |
| return std::make_unique<TimestampNsType>(); | |
| } else if (normalized_type_name == "timestamptz_ns") { | |
| return std::make_unique<TimestampTzNsType>(); | |
| } else if (normalized_type_name == "string") { | |
| return std::make_unique<StringType>(); | |
| } else if (normalized_type_name == "binary") { | |
| return std::make_unique<BinaryType>(); | |
| } else if (normalized_type_name == "uuid") { | |
| return std::make_unique<UuidType>(); | |
| } else if (normalized_type_name == "unknown") { | |
| return std::make_unique<UnknownType>(); | |
| } else if (normalized_type_name == "variant") { | |
| return std::make_unique<VariantType>(); | |
| } else if (normalized_type_name.starts_with("fixed")) { | |
| static const std::regex kFixedRegex(R"(fixed\[\s*(\d+)\s*\])"); | |
| std::smatch match; | |
| if (std::regex_match(normalized_type_name, match, kFixedRegex)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto length, | |
| StringUtils::ParseNumber<int32_t>(match[1].str())); | |
| return std::make_unique<FixedType>(length); | |
| } | |
| return JsonParseError("Invalid fixed type: {}", type_name); | |
| } else if (normalized_type_name.starts_with("decimal")) { | |
| static const std::regex kDecimalRegex(R"(decimal\(\s*(\d+)\s*,\s*(\d+)\s*\))"); | |
| std::smatch match; | |
| if (std::regex_match(normalized_type_name, match, kDecimalRegex)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto precision, | |
| StringUtils::ParseNumber<int32_t>(match[1].str())); | |
| ICEBERG_ASSIGN_OR_RAISE(auto scale, | |
| StringUtils::ParseNumber<int32_t>(match[2].str())); | |
| return std::make_unique<DecimalType>(precision, scale); | |
| } | |
| return JsonParseError("Invalid decimal type: {}", type_name); | |
| } else if (normalized_type_name.starts_with("geometry")) { | |
| static const std::regex kGeometryRegex(R"(geometry\s*(?:\(\s*([^)]*?)\s*\))?)", | |
| std::regex_constants::icase); | |
| std::smatch match; | |
| if (std::regex_match(type_name, match, kGeometryRegex)) { | |
| if (match[1].matched) { | |
| auto crs = match[1].str(); | |
| if (crs.empty()) { | |
| return JsonParseError("Invalid geometry type: {}", type_name); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, GeometryType::Make(std::move(crs))); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, GeometryType::Make()); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| return JsonParseError("Invalid geometry type: {}", type_name); | |
| } else if (normalized_type_name.starts_with("geography")) { | |
| static const std::regex kGeographyRegex( | |
| R"(geography\s*(?:\(\s*([^,]*?)\s*(?:,\s*(\w*)\s*)?\))?)", | |
| std::regex_constants::icase); | |
| std::smatch match; | |
| if (std::regex_match(type_name, match, kGeographyRegex)) { | |
| auto crs = match[1].str(); | |
| if (match[1].matched && crs.empty()) { | |
| return JsonParseError("Invalid geography type: {}", type_name); | |
| } | |
| if (match[2].matched) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto algorithm, | |
| EdgeAlgorithmFromString(match[2].str())); | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, | |
| GeographyType::Make(std::move(crs), algorithm)); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| if (match[1].matched) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, GeographyType::Make(std::move(crs))); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, GeographyType::Make()); | |
| return std::unique_ptr<Type>(std::move(type)); | |
| } | |
| return JsonParseError("Invalid geography type: {}", type_name); | |
| } else { | |
| return JsonParseError("Cannot parse type string: {}", type_name); | |
| } | |
| } | |
| // For complex types like struct, list, and map | |
| ICEBERG_ASSIGN_OR_RAISE(auto complex_type_name, GetJsonValue<std::string>(json, kType)); | |
| if (complex_type_name == kStruct) { | |
| return StructTypeFromJson(json); | |
| } else if (complex_type_name == kList) { | |
| return ListTypeFromJson(json); | |
| } else if (complex_type_name == kMap) { | |
| return MapTypeFromJson(json); | |
| } else { | |
| return JsonParseError("Unknown complex type: {}", complex_type_name); | |
| } | |
| } | |
| namespace { | |
| // The spec's JSON single-value form for `timestamptz` / `timestamptz_ns` default | |
| // values requires a UTC offset. The shared timestamp parser accepts any offset and | |
| // silently normalizes to UTC, which would let C++ accept default metadata that Java | |
| // rejects and then rewrite the offset on serialization. Enforce UTC for these | |
| // defaults at parse time, where the original offset is still visible. | |
| Status ValidateTimestamptzDefaultIsUtc(const Type& type, const nlohmann::json& value) { | |
| const auto type_id = type.type_id(); | |
| if (type_id != TypeId::kTimestampTz && type_id != TypeId::kTimestampTzNs) { | |
| return {}; | |
| } | |
| if (!value.is_string()) { | |
| return JsonParseError( | |
| "Invalid timestamptz default for {}: expected a string (JSON value of type {})", | |
| type.ToString(), value.type_name()); | |
| } | |
| const auto str = value.get<std::string>(); | |
| ICEBERG_ASSIGN_OR_RAISE(bool is_utc, TemporalUtils::IsUtcOffset(str)); | |
| if (!is_utc) { | |
| return JsonParseError( | |
| "Invalid timestamptz default '{}' for {}: default values must use a UTC offset", | |
| str, type.ToString()); | |
| } | |
| return {}; | |
| } | |
| } // namespace | |
| Result<std::unique_ptr<SchemaField>> FieldFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto type, GetJsonValue<nlohmann::json>(json, kType).and_then(TypeFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto field_id, GetJsonInteger<int32_t>(json, kId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto name, GetJsonValue<std::string>(json, kName)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto required, GetJsonValue<bool>(json, kRequired)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto doc, GetJsonValueOrDefault<std::string>(json, kDoc)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto initial_default_json, | |
| GetJsonValueOptional<nlohmann::json>(json, kInitialDefault)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto write_default_json, | |
| GetJsonValueOptional<nlohmann::json>(json, kWriteDefault)); | |
| std::shared_ptr<const Literal> initial_default; | |
| if (initial_default_json.has_value()) { | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ValidateTimestamptzDefaultIsUtc(*type, *initial_default_json)); | |
| ICEBERG_ASSIGN_OR_RAISE(Literal literal, | |
| LiteralFromJson(*initial_default_json, type.get())); | |
| initial_default = std::make_shared<const Literal>(std::move(literal)); | |
| } | |
| std::shared_ptr<const Literal> write_default; | |
| if (write_default_json.has_value()) { | |
| ICEBERG_RETURN_UNEXPECTED( | |
| ValidateTimestamptzDefaultIsUtc(*type, *write_default_json)); | |
| ICEBERG_ASSIGN_OR_RAISE(Literal literal, | |
| LiteralFromJson(*write_default_json, type.get())); | |
| write_default = std::make_shared<const Literal>(std::move(literal)); | |
| } | |
| return std::make_unique<SchemaField>(field_id, std::move(name), std::move(type), | |
| !required, doc, std::move(initial_default), | |
| std::move(write_default)); | |
| } | |
| Result<std::unique_ptr<Schema>> SchemaFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_id_opt, | |
| GetJsonIntegerOptional<int32_t>(json, kSchemaId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, TypeFromJson(json)); | |
| if (type->type_id() != TypeId::kStruct) [[unlikely]] { | |
| return JsonParseError("Schema must be a struct type, but got {}", type->ToString()); | |
| } | |
| auto& struct_type = internal::checked_cast<StructType&>(*type); | |
| std::vector<SchemaField> fields; | |
| fields.reserve(struct_type.fields().size()); | |
| for (auto& field : struct_type.fields()) { | |
| fields.emplace_back(std::move(field)); | |
| } | |
| std::vector<int32_t> identifier_field_ids; | |
| if (json.contains(kIdentifierFieldIds) && !json.at(kIdentifierFieldIds).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(identifier_field_ids, | |
| IntegerVectorFromJson<int32_t>(json, kIdentifierFieldIds)); | |
| } | |
| return Schema::Make(std::move(fields), schema_id_opt.value_or(Schema::kInitialSchemaId), | |
| std::move(identifier_field_ids)); | |
| } | |
| nlohmann::json ToJson(const PartitionField& partition_field) { | |
| nlohmann::json json; | |
| json[kSourceId] = partition_field.source_id(); | |
| json[kFieldId] = partition_field.field_id(); | |
| json[kTransform] = std::format("{}", *partition_field.transform()); | |
| json[kName] = partition_field.name(); | |
| return json; | |
| } | |
| nlohmann::json PartitionFieldsToJson(const PartitionSpec& partition_spec) { | |
| nlohmann::json fields_json = nlohmann::json::array(); | |
| for (const auto& field : partition_spec.fields()) { | |
| fields_json.push_back(ToJson(field)); | |
| } | |
| return fields_json; | |
| } | |
| Result<std::string> PartitionFieldsToJsonString(const PartitionSpec& partition_spec) { | |
| return ToJsonString(PartitionFieldsToJson(partition_spec)); | |
| } | |
| nlohmann::json ToJson(const PartitionSpec& partition_spec) { | |
| nlohmann::json json; | |
| json[kSpecId] = partition_spec.spec_id(); | |
| json[kFields] = PartitionFieldsToJson(partition_spec); | |
| return json; | |
| } | |
| Result<std::string> ToJsonString(const PartitionSpec& partition_spec) { | |
| return ToJsonString(ToJson(partition_spec)); | |
| } | |
| Result<std::unique_ptr<PartitionField>> PartitionFieldFromJson( | |
| const nlohmann::json& json, bool allow_field_id_missing) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto source_id, GetJsonInteger<int32_t>(json, kSourceId)); | |
| int32_t field_id; | |
| if (allow_field_id_missing && !json.contains(kFieldId)) { | |
| // Partition field id in v1 is not tracked, so we use -1 to indicate that. | |
| field_id = SchemaField::kInvalidFieldId; | |
| } else { | |
| ICEBERG_ASSIGN_OR_RAISE(field_id, GetJsonInteger<int32_t>(json, kFieldId)); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto transform, | |
| GetJsonValue<std::string>(json, kTransform).and_then(TransformFromString)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto name, GetJsonValue<std::string>(json, kName)); | |
| return std::make_unique<PartitionField>(source_id, field_id, name, | |
| std::move(transform)); | |
| } | |
| namespace { | |
| Result<std::unique_ptr<PartitionSpec>> ParseBoundPartitionSpec( | |
| const Schema& schema, const nlohmann::json& json, int32_t spec_id, | |
| bool allow_missing_fields) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto fields, GetJsonValue<nlohmann::json>(json, kFields)); | |
| std::vector<PartitionField> partition_fields; | |
| for (const auto& field_json : fields) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_field, PartitionFieldFromJson(field_json)); | |
| partition_fields.push_back(std::move(*partition_field)); | |
| } | |
| return PartitionSpec::Make(schema, spec_id, std::move(partition_fields), | |
| allow_missing_fields); | |
| } | |
| } // namespace | |
| Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson( | |
| const std::shared_ptr<Schema>& schema, const nlohmann::json& json, | |
| int32_t default_spec_id) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonInteger<int32_t>(json, kSpecId)); | |
| return ParseBoundPartitionSpec(*schema, json, spec_id, | |
| /*allow_missing_fields=*/spec_id != default_spec_id); | |
| } | |
| Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonInteger<int32_t>(json, kSpecId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto fields, GetJsonValue<nlohmann::json>(json, kFields)); | |
| std::vector<PartitionField> partition_fields; | |
| for (const auto& field_json : fields) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_field, PartitionFieldFromJson(field_json)); | |
| partition_fields.push_back(std::move(*partition_field)); | |
| } | |
| return PartitionSpec::Make(spec_id, std::move(partition_fields)); | |
| } | |
| Result<std::unique_ptr<SnapshotRef>> SnapshotRefFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto type, | |
| GetJsonValue<std::string>(json, kType).and_then(SnapshotRefTypeFromString)); | |
| if (type == SnapshotRefType::kBranch) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto min_snapshots_to_keep, | |
| GetJsonValueOptional<int32_t>(json, kMinSnapshotsToKeep)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto max_snapshot_age_ms, | |
| GetJsonValueOptional<int64_t>(json, kMaxSnapshotAgeMs)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto max_ref_age_ms, | |
| GetJsonValueOptional<int64_t>(json, kMaxRefAgeMs)); | |
| return std::make_unique<SnapshotRef>( | |
| snapshot_id, SnapshotRef::Branch{.min_snapshots_to_keep = min_snapshots_to_keep, | |
| .max_snapshot_age_ms = max_snapshot_age_ms, | |
| .max_ref_age_ms = max_ref_age_ms}); | |
| } else { | |
| ICEBERG_ASSIGN_OR_RAISE(auto max_ref_age_ms, | |
| GetJsonValueOptional<int64_t>(json, kMaxRefAgeMs)); | |
| return std::make_unique<SnapshotRef>( | |
| snapshot_id, SnapshotRef::Tag{.max_ref_age_ms = max_ref_age_ms}); | |
| } | |
| } | |
| Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto sequence_number, | |
| GetJsonValueOptional<int64_t>(json, kSequenceNumber)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs)); | |
| auto timestamp_ms = TimePointMsFromUnixMs(unix_ms); | |
| ICEBERG_ASSIGN_OR_RAISE(auto manifest_list, | |
| GetJsonValue<std::string>(json, kManifestList)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto parent_snapshot_id, | |
| GetJsonValueOptional<int64_t>(json, kParentSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto summary_json, | |
| GetJsonValueOptional<nlohmann::json>(json, kSummary)); | |
| std::unordered_map<std::string, std::string> summary; | |
| if (summary_json.has_value()) { | |
| for (const auto& [key, value] : summary_json->items()) { | |
| // if (!kValidSnapshotSummaryFields.contains(key)) { | |
| // return JsonParseError("Invalid snapshot summary field: {}", key); | |
| // } | |
| if (!value.is_string()) { | |
| return JsonParseError( | |
| "Invalid snapshot summary field value for '{}' (JSON value of type {})", key, | |
| value.type_name()); | |
| } | |
| if (key == SnapshotSummaryFields::kOperation && | |
| !kValidDataOperation.contains(value.get<std::string>())) { | |
| return JsonParseError("Invalid snapshot operation in snapshot summary"); | |
| } | |
| summary[key] = value.get<std::string>(); | |
| } | |
| // If summary is available but operation is missing, set operation to overwrite. | |
| if (!summary.contains(SnapshotSummaryFields::kOperation)) { | |
| summary[SnapshotSummaryFields::kOperation] = DataOperation::kOverwrite; | |
| } | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto first_row_id, | |
| GetJsonValueOptional<int64_t>(json, kFirstRowId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto added_rows, | |
| GetJsonValueOptional<int64_t>(json, kAddedRows)); | |
| if (first_row_id.has_value() && first_row_id.value() < 0) { | |
| return JsonParseError("Invalid first-row-id (cannot be negative): {}", | |
| first_row_id.value()); | |
| } | |
| if (added_rows.has_value() && added_rows.value() < 0) { | |
| return JsonParseError("Invalid added-rows (cannot be negative): {}", | |
| added_rows.value()); | |
| } | |
| if (first_row_id.has_value() && !added_rows.has_value()) { | |
| return JsonParseError("Invalid added-rows (required when first-row-id is set): null"); | |
| } | |
| if (!first_row_id.has_value()) { | |
| added_rows = std::nullopt; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValueOptional<int32_t>(json, kSchemaId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValueOptional<std::string>(json, kKeyId)); | |
| return std::make_unique<Snapshot>( | |
| snapshot_id, parent_snapshot_id, | |
| sequence_number.value_or(TableMetadata::kInitialSequenceNumber), timestamp_ms, | |
| manifest_list, std::move(summary), schema_id, first_row_id, added_rows, | |
| std::move(key_id)); | |
| } | |
| nlohmann::json ToJson(const BlobMetadata& blob_metadata) { | |
| nlohmann::json json; | |
| json[kType] = blob_metadata.type; | |
| json[kSnapshotId] = blob_metadata.source_snapshot_id; | |
| json[kSequenceNumber] = blob_metadata.source_snapshot_sequence_number; | |
| json[kFields] = blob_metadata.fields; | |
| if (!blob_metadata.properties.empty()) { | |
| json[kProperties] = blob_metadata.properties; | |
| } | |
| return json; | |
| } | |
| Result<BlobMetadata> BlobMetadataFromJson(const nlohmann::json& json) { | |
| BlobMetadata blob_metadata; | |
| ICEBERG_ASSIGN_OR_RAISE(blob_metadata.type, GetJsonValue<std::string>(json, kType)); | |
| ICEBERG_ASSIGN_OR_RAISE(blob_metadata.source_snapshot_id, | |
| GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE(blob_metadata.source_snapshot_sequence_number, | |
| GetJsonValue<int64_t>(json, kSequenceNumber)); | |
| ICEBERG_ASSIGN_OR_RAISE(blob_metadata.fields, | |
| GetJsonValue<std::vector<int32_t>>(json, kFields)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| blob_metadata.properties, | |
| (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json, | |
| kProperties))); | |
| return blob_metadata; | |
| } | |
| nlohmann::json ToJson(const StatisticsFile& statistics_file) { | |
| nlohmann::json json; | |
| json[kSnapshotId] = statistics_file.snapshot_id; | |
| json[kStatisticsPath] = statistics_file.path; | |
| json[kFileSizeInBytes] = statistics_file.file_size_in_bytes; | |
| json[kFileFooterSizeInBytes] = statistics_file.file_footer_size_in_bytes; | |
| nlohmann::json blob_metadata_array = nlohmann::json::array(); | |
| for (const auto& blob_metadata : statistics_file.blob_metadata) { | |
| blob_metadata_array.push_back(ToJson(blob_metadata)); | |
| } | |
| json[kBlobMetadata] = blob_metadata_array; | |
| return json; | |
| } | |
| Result<std::unique_ptr<StatisticsFile>> StatisticsFileFromJson( | |
| const nlohmann::json& json) { | |
| auto stats_file = std::make_unique<StatisticsFile>(); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->snapshot_id, | |
| GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->path, | |
| GetJsonValue<std::string>(json, kStatisticsPath)); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->file_size_in_bytes, | |
| GetJsonValue<int64_t>(json, kFileSizeInBytes)); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->file_footer_size_in_bytes, | |
| GetJsonValue<int64_t>(json, kFileFooterSizeInBytes)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto blob_metadata_array, | |
| GetJsonValue<nlohmann::json>(json, kBlobMetadata)); | |
| for (const auto& blob_json : blob_metadata_array) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto blob, BlobMetadataFromJson(blob_json)); | |
| stats_file->blob_metadata.push_back(std::move(blob)); | |
| } | |
| return stats_file; | |
| } | |
| nlohmann::json ToJson(const PartitionStatisticsFile& partition_statistics_file) { | |
| nlohmann::json json; | |
| json[kSnapshotId] = partition_statistics_file.snapshot_id; | |
| json[kStatisticsPath] = partition_statistics_file.path; | |
| json[kFileSizeInBytes] = partition_statistics_file.file_size_in_bytes; | |
| return json; | |
| } | |
| Result<std::unique_ptr<PartitionStatisticsFile>> PartitionStatisticsFileFromJson( | |
| const nlohmann::json& json) { | |
| auto stats_file = std::make_unique<PartitionStatisticsFile>(); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->snapshot_id, | |
| GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->path, | |
| GetJsonValue<std::string>(json, kStatisticsPath)); | |
| ICEBERG_ASSIGN_OR_RAISE(stats_file->file_size_in_bytes, | |
| GetJsonValue<int64_t>(json, kFileSizeInBytes)); | |
| return stats_file; | |
| } | |
| nlohmann::json ToJson(const SnapshotLogEntry& snapshot_log_entry) { | |
| nlohmann::json json; | |
| json[kTimestampMs] = UnixMsFromTimePointMs(snapshot_log_entry.timestamp_ms); | |
| json[kSnapshotId] = snapshot_log_entry.snapshot_id; | |
| return json; | |
| } | |
| Result<SnapshotLogEntry> SnapshotLogEntryFromJson(const nlohmann::json& json) { | |
| SnapshotLogEntry snapshot_log_entry; | |
| ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs)); | |
| snapshot_log_entry.timestamp_ms = TimePointMsFromUnixMs(unix_ms); | |
| ICEBERG_ASSIGN_OR_RAISE(snapshot_log_entry.snapshot_id, | |
| GetJsonValue<int64_t>(json, kSnapshotId)); | |
| return snapshot_log_entry; | |
| } | |
| nlohmann::json ToJson(const MetadataLogEntry& metadata_log_entry) { | |
| nlohmann::json json; | |
| json[kTimestampMs] = UnixMsFromTimePointMs(metadata_log_entry.timestamp_ms); | |
| json[kMetadataFile] = metadata_log_entry.metadata_file; | |
| return json; | |
| } | |
| Result<MetadataLogEntry> MetadataLogEntryFromJson(const nlohmann::json& json) { | |
| MetadataLogEntry metadata_log_entry; | |
| ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue<int64_t>(json, kTimestampMs)); | |
| metadata_log_entry.timestamp_ms = TimePointMsFromUnixMs(unix_ms); | |
| ICEBERG_ASSIGN_OR_RAISE(metadata_log_entry.metadata_file, | |
| GetJsonValue<std::string>(json, kMetadataFile)); | |
| return metadata_log_entry; | |
| } | |
| nlohmann::json ToJson(const EncryptedKey& encrypted_key) { | |
| nlohmann::json json; | |
| json[kKeyId] = encrypted_key.key_id; | |
| json[kEncryptedKeyMetadata] = Base64::Encode(encrypted_key.encrypted_key_metadata); | |
| SetOptionalField(json, kEncryptedById, encrypted_key.encrypted_by_id); | |
| if (!encrypted_key.properties.empty()) { | |
| json[kProperties] = encrypted_key.properties; | |
| } | |
| return json; | |
| } | |
| Result<EncryptedKey> EncryptedKeyFromJson(const nlohmann::json& json) { | |
| using StringMap = std::unordered_map<std::string, std::string>; | |
| if (!json.is_object()) { | |
| return JsonParseError( | |
| "Invalid encryption key, must be non-null object (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValue<std::string>(json, kKeyId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto encoded_metadata, | |
| GetJsonValue<std::string>(json, kEncryptedKeyMetadata)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto encrypted_key_metadata, Base64::Decode(encoded_metadata)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto encrypted_by_id, | |
| GetJsonValueOptional<std::string>(json, kEncryptedById)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto properties, | |
| GetJsonValueOrDefault<StringMap>(json, kProperties)); | |
| return EncryptedKey{ | |
| .key_id = std::move(key_id), | |
| .encrypted_key_metadata = std::move(encrypted_key_metadata), | |
| .encrypted_by_id = std::move(encrypted_by_id), | |
| .properties = std::move(properties), | |
| }; | |
| } | |
| Result<nlohmann::json> ToJson(const TableMetadata& table_metadata) { | |
| nlohmann::json json; | |
| json[kFormatVersion] = table_metadata.format_version; | |
| json[kTableUuid] = table_metadata.table_uuid; | |
| json[kLocation] = table_metadata.location; | |
| if (table_metadata.format_version > 1) { | |
| json[kLastSequenceNumber] = table_metadata.last_sequence_number; | |
| } | |
| json[kLastUpdatedMs] = UnixMsFromTimePointMs(table_metadata.last_updated_ms); | |
| json[kLastColumnId] = table_metadata.last_column_id; | |
| // for older readers, continue writing the current schema as "schema". | |
| // this is only needed for v1 because support for schemas and current-schema-id | |
| // is required in v2 and later. | |
| if (table_metadata.format_version == 1) { | |
| for (const auto& schema : table_metadata.schemas) { | |
| if (schema->schema_id() == table_metadata.current_schema_id) { | |
| ICEBERG_ASSIGN_OR_RAISE(json[kSchema], ToJson(*schema)); | |
| break; | |
| } | |
| } | |
| } | |
| // write the current schema ID and schema list | |
| json[kCurrentSchemaId] = table_metadata.current_schema_id; | |
| // ToJson(Schema) is fallible, so the shared ToJsonList helper (which assumes an | |
| // infallible ToJson) cannot be used here; build the array with an explicit loop. | |
| nlohmann::json schemas_json = nlohmann::json::array(); | |
| for (const auto& schema : table_metadata.schemas) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_json, ToJson(*schema)); | |
| schemas_json.push_back(std::move(schema_json)); | |
| } | |
| json[kSchemas] = std::move(schemas_json); | |
| // for older readers, continue writing the default spec as "partition-spec" | |
| if (table_metadata.format_version == 1) { | |
| for (const auto& partition_spec : table_metadata.partition_specs) { | |
| if (partition_spec->spec_id() == table_metadata.default_spec_id) { | |
| json[kPartitionSpec] = ToJson(*partition_spec); | |
| break; | |
| } | |
| } | |
| } | |
| // write the default spec ID and spec list | |
| json[kDefaultSpecId] = table_metadata.default_spec_id; | |
| json[kPartitionSpecs] = ToJsonList(table_metadata.partition_specs); | |
| json[kLastPartitionId] = table_metadata.last_partition_id; | |
| // write the default order ID and sort order list | |
| json[kDefaultSortOrderId] = table_metadata.default_sort_order_id; | |
| json[kSortOrders] = ToJsonList(table_metadata.sort_orders); | |
| // write properties map | |
| json[kProperties] = table_metadata.properties.configs(); | |
| if (std::ranges::any_of(table_metadata.snapshots, [&](const auto& snapshot) { | |
| return snapshot->snapshot_id == table_metadata.current_snapshot_id; | |
| })) { | |
| json[kCurrentSnapshotId] = table_metadata.current_snapshot_id; | |
| } else { | |
| json[kCurrentSnapshotId] = nlohmann::json::value_t::null; | |
| } | |
| if (table_metadata.format_version >= 3) { | |
| json[kNextRowId] = table_metadata.next_row_id; | |
| } | |
| json[kRefs] = ToJsonMap(table_metadata.refs); | |
| json[kSnapshots] = ToJsonList(table_metadata.snapshots); | |
| json[kStatistics] = ToJsonList(table_metadata.statistics); | |
| json[kPartitionStatistics] = ToJsonList(table_metadata.partition_statistics); | |
| if (!table_metadata.encryption_keys.empty()) { | |
| json[kEncryptionKeys] = ToJsonList(table_metadata.encryption_keys); | |
| } | |
| json[kSnapshotLog] = ToJsonList(table_metadata.snapshot_log); | |
| json[kMetadataLog] = ToJsonList(table_metadata.metadata_log); | |
| return json; | |
| } | |
| Result<std::string> ToJsonString(const TableMetadata& table_metadata) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto json, ToJson(table_metadata)); | |
| return ToJsonString(json); | |
| } | |
| namespace { | |
| /// \brief Parse the schemas from the JSON object. | |
| /// | |
| /// \param[in] json The JSON object to parse. | |
| /// \param[in] format_version The format version of the table. | |
| /// \param[out] current_schema_id The current schema ID. | |
| /// \param[out] schemas The list of schemas. | |
| /// | |
| /// \return The current schema or parse error. | |
| Result<std::shared_ptr<Schema>> ParseSchemas( | |
| const nlohmann::json& json, int8_t format_version, int32_t& current_schema_id, | |
| std::vector<std::shared_ptr<Schema>>& schemas) { | |
| std::shared_ptr<Schema> current_schema; | |
| if (json.contains(kSchemas)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_array, | |
| GetJsonValue<nlohmann::json>(json, kSchemas)); | |
| if (!schema_array.is_array()) { | |
| return JsonParseError("Cannot parse schemas from non-array (JSON value of type {})", | |
| schema_array.type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(current_schema_id, | |
| GetJsonInteger<int32_t>(json, kCurrentSchemaId)); | |
| for (const auto& schema_json : schema_array) { | |
| ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<Schema> schema, | |
| SchemaFromJson(schema_json)); | |
| ICEBERG_RETURN_UNEXPECTED(schema->Validate(format_version)); | |
| if (schema->schema_id() == current_schema_id) { | |
| current_schema = schema; | |
| } | |
| schemas.push_back(std::move(schema)); | |
| } | |
| if (!current_schema) { | |
| return JsonParseError("Cannot find schema with {}={}", kCurrentSchemaId, | |
| current_schema_id); | |
| } | |
| } else { | |
| if (format_version != 1) { | |
| return JsonParseError("{} must exist in format v{}", kSchemas, format_version); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_json, | |
| GetJsonValue<nlohmann::json>(json, kSchema)); | |
| ICEBERG_ASSIGN_OR_RAISE(current_schema, SchemaFromJson(schema_json)); | |
| ICEBERG_RETURN_UNEXPECTED(current_schema->Validate(format_version)); | |
| current_schema_id = current_schema->schema_id(); | |
| schemas.push_back(current_schema); | |
| } | |
| return current_schema; | |
| } | |
| /// \brief Parse the partition specs from the JSON object. | |
| /// | |
| /// \param[in] json The JSON object to parse. | |
| /// \param[in] format_version The format version of the table. | |
| /// \param[in] current_schema The current schema. | |
| /// \param[out] default_spec_id The default partition spec ID. | |
| /// \param[out] partition_specs The list of partition specs. | |
| Status ParsePartitionSpecs(const nlohmann::json& json, int8_t format_version, | |
| const std::shared_ptr<Schema>& current_schema, | |
| int32_t& default_spec_id, | |
| std::vector<std::shared_ptr<PartitionSpec>>& partition_specs) { | |
| if (json.contains(kPartitionSpecs)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_array, | |
| GetJsonValue<nlohmann::json>(json, kPartitionSpecs)); | |
| if (!spec_array.is_array()) { | |
| return JsonParseError( | |
| "Cannot parse partition specs from non-array (JSON value of type {})", | |
| spec_array.type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(default_spec_id, | |
| GetJsonInteger<int32_t>(json, kDefaultSpecId)); | |
| for (const auto& spec_json : spec_array) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto spec, PartitionSpecFromJson(current_schema, spec_json, default_spec_id)); | |
| partition_specs.push_back(std::move(spec)); | |
| } | |
| } else { | |
| if (format_version != 1) { | |
| return JsonParseError("{} must exist in format v{}", kPartitionSpecs, | |
| format_version); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json, | |
| GetJsonValue<nlohmann::json>(json, kPartitionSpec)); | |
| if (!partition_spec_json.is_array()) { | |
| return JsonParseError( | |
| "Cannot parse v1 partition spec from non-array (JSON value of type {})", | |
| partition_spec_json.type_name()); | |
| } | |
| int32_t next_partition_field_id = PartitionSpec::kLegacyPartitionDataIdStart; | |
| std::vector<PartitionField> fields; | |
| for (const auto& entry_json : partition_spec_json) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto field, PartitionFieldFromJson( | |
| entry_json, /*allow_field_id_missing=*/format_version == 1)); | |
| int32_t field_id = field->field_id(); | |
| if (field_id == SchemaField::kInvalidFieldId) { | |
| // If the field ID is not set, we need to assign a new one | |
| field_id = next_partition_field_id++; | |
| } | |
| fields.emplace_back(field->source_id(), field_id, std::string(field->name()), | |
| std::move(field->transform())); | |
| } | |
| // Create partition spec with schema validation | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto spec, | |
| PartitionSpec::Make(*current_schema, PartitionSpec::kInitialSpecId, | |
| std::move(fields), /*allow_missing_fields=*/false)); | |
| default_spec_id = spec->spec_id(); | |
| partition_specs.push_back(std::move(spec)); | |
| } | |
| return {}; | |
| } | |
| /// \brief Parse the sort orders from the JSON object. | |
| /// | |
| /// \param[in] json The JSON object to parse. | |
| /// \param[in] format_version The format version of the table. | |
| /// \param[in] current_schema The current schema. | |
| /// \param[out] default_sort_order_id The default sort order ID. | |
| /// \param[out] sort_orders The list of sort orders. | |
| Status ParseSortOrders(const nlohmann::json& json, int8_t format_version, | |
| const std::shared_ptr<Schema>& current_schema, | |
| int32_t& default_sort_order_id, | |
| std::vector<std::shared_ptr<SortOrder>>& sort_orders) { | |
| if (json.contains(kSortOrders)) { | |
| ICEBERG_ASSIGN_OR_RAISE(default_sort_order_id, | |
| GetJsonValue<int32_t>(json, kDefaultSortOrderId)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_order_array, | |
| GetJsonValue<nlohmann::json>(json, kSortOrders)); | |
| for (const auto& sort_order_json : sort_order_array) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto sort_order, | |
| SortOrderFromJson(sort_order_json, current_schema, default_sort_order_id)); | |
| sort_orders.push_back(std::move(sort_order)); | |
| } | |
| } else { | |
| if (format_version > 1) { | |
| return JsonParseError("{} must exist in format v{}", kSortOrders, format_version); | |
| } | |
| auto sort_order = SortOrder::Unsorted(); | |
| default_sort_order_id = sort_order->order_id(); | |
| sort_orders.push_back(std::move(sort_order)); | |
| } | |
| return {}; | |
| } | |
| } // namespace | |
| Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::json& json) { | |
| if (!json.is_object()) { | |
| return JsonParseError( | |
| "Cannot parse metadata from a non-object (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| auto table_metadata = std::make_unique<TableMetadata>(); | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->format_version, | |
| GetJsonValue<int8_t>(json, kFormatVersion)); | |
| if (table_metadata->format_version < 1 || | |
| table_metadata->format_version > TableMetadata::kSupportedTableFormatVersion) { | |
| return JsonParseError("Cannot read unsupported version: {}", | |
| table_metadata->format_version); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->table_uuid, | |
| GetJsonValueOrDefault<std::string>(json, kTableUuid)); | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->location, | |
| GetJsonValue<std::string>(json, kLocation)); | |
| if (table_metadata->format_version > 1) { | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_sequence_number, | |
| GetJsonValue<int64_t>(json, kLastSequenceNumber)); | |
| } else { | |
| table_metadata->last_sequence_number = TableMetadata::kInitialSequenceNumber; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_column_id, | |
| GetJsonValue<int32_t>(json, kLastColumnId)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto current_schema, | |
| ParseSchemas(json, table_metadata->format_version, | |
| table_metadata->current_schema_id, table_metadata->schemas)); | |
| ICEBERG_RETURN_UNEXPECTED(ParsePartitionSpecs( | |
| json, table_metadata->format_version, current_schema, | |
| table_metadata->default_spec_id, table_metadata->partition_specs)); | |
| if (json.contains(kLastPartitionId)) { | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_partition_id, | |
| GetJsonValue<int32_t>(json, kLastPartitionId)); | |
| } else { | |
| if (table_metadata->format_version > 1) { | |
| return JsonParseError("{} must exist in format v{}", kLastPartitionId, | |
| table_metadata->format_version); | |
| } | |
| if (table_metadata->partition_specs.empty()) { | |
| table_metadata->last_partition_id = | |
| PartitionSpec::Unpartitioned()->last_assigned_field_id(); | |
| } else { | |
| table_metadata->last_partition_id = | |
| std::ranges::max(table_metadata->partition_specs, {}, [](const auto& spec) { | |
| return spec->last_assigned_field_id(); | |
| })->last_assigned_field_id(); | |
| } | |
| } | |
| ICEBERG_RETURN_UNEXPECTED(ParseSortOrders( | |
| json, table_metadata->format_version, current_schema, | |
| table_metadata->default_sort_order_id, table_metadata->sort_orders)); | |
| if (json.contains(kProperties)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto properties, FromJsonMap(json, kProperties)); | |
| table_metadata->properties = TableProperties::FromMap(std::move(properties)); | |
| } | |
| // This field is optional, but internally we set this to -1 when not set | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->current_snapshot_id, | |
| GetJsonValueOrDefault<int64_t>(json, kCurrentSnapshotId, kInvalidSnapshotId)); | |
| if (table_metadata->format_version >= 3) { | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->next_row_id, | |
| GetJsonValue<int64_t>(json, kNextRowId)); | |
| } else { | |
| table_metadata->next_row_id = TableMetadata::kInitialRowId; | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto last_updated_ms, | |
| GetJsonValue<int64_t>(json, kLastUpdatedMs)); | |
| table_metadata->last_updated_ms = | |
| TimePointMs{std::chrono::milliseconds(last_updated_ms)}; | |
| if (json.contains(kRefs)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto refs, FromJsonMap<std::shared_ptr<SnapshotRef>>( | |
| json, kRefs, SnapshotRefFromJson)); | |
| table_metadata->refs = std::move(refs); | |
| } else if (table_metadata->current_snapshot_id != kInvalidSnapshotId) { | |
| table_metadata->refs["main"] = std::make_unique<SnapshotRef>(SnapshotRef{ | |
| .snapshot_id = table_metadata->current_snapshot_id, | |
| .retention = SnapshotRef::Branch{}, | |
| }); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(table_metadata->snapshots, | |
| FromJsonList<Snapshot>(json, kSnapshots, SnapshotFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->statistics, | |
| FromJsonList<StatisticsFile>(json, kStatistics, StatisticsFileFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->partition_statistics, | |
| FromJsonList<PartitionStatisticsFile>(json, kPartitionStatistics, | |
| PartitionStatisticsFileFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->encryption_keys, | |
| FromJsonList<EncryptedKey>(json, kEncryptionKeys, EncryptedKeyFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->snapshot_log, | |
| FromJsonList<SnapshotLogEntry>(json, kSnapshotLog, SnapshotLogEntryFromJson)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| table_metadata->metadata_log, | |
| FromJsonList<MetadataLogEntry>(json, kMetadataLog, MetadataLogEntryFromJson)); | |
| return table_metadata; | |
| } | |
| Result<nlohmann::json> FromJsonString(const std::string& json_string) { | |
| return ParseJsonString(json_string); | |
| } | |
| Result<std::string> ToJsonString(const nlohmann::json& json) { | |
| try { | |
| return json.dump(); | |
| } catch (const std::exception& e) { | |
| return JsonParseError("Failed to serialize to JSON string: {}", e.what()); | |
| } | |
| } | |
| nlohmann::json ToJson(const MappedField& field) { | |
| nlohmann::json json; | |
| if (field.field_id.has_value()) { | |
| json[kFieldId] = field.field_id.value(); | |
| } | |
| nlohmann::json names = nlohmann::json::array(); | |
| for (const auto& name : field.names) { | |
| names.push_back(name); | |
| } | |
| json[kNames] = names; | |
| if (field.nested_mapping != nullptr) { | |
| json[kFields] = ToJson(*field.nested_mapping); | |
| } | |
| return json; | |
| } | |
| Result<MappedField> MappedFieldFromJson(const nlohmann::json& json) { | |
| if (!json.is_object()) [[unlikely]] { | |
| return JsonParseError("Cannot parse non-object mapping field (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(std::optional<int32_t> field_id, | |
| GetJsonValueOptional<int32_t>(json, kFieldId)); | |
| std::vector<std::string> names; | |
| if (json.contains(kNames)) { | |
| ICEBERG_ASSIGN_OR_RAISE(names, GetJsonValue<std::vector<std::string>>(json, kNames)); | |
| } | |
| std::unique_ptr<MappedFields> nested_mapping; | |
| if (json.contains(kFields)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto fields_json, | |
| GetJsonValue<nlohmann::json>(json, kFields)); | |
| ICEBERG_ASSIGN_OR_RAISE(nested_mapping, MappedFieldsFromJson(fields_json)); | |
| } | |
| return MappedField{.names = {names.cbegin(), names.cend()}, | |
| .field_id = field_id, | |
| .nested_mapping = std::move(nested_mapping)}; | |
| } | |
| nlohmann::json ToJson(const MappedFields& mapped_fields) { | |
| nlohmann::json array = nlohmann::json::array(); | |
| for (const auto& field : mapped_fields.fields()) { | |
| array.push_back(ToJson(field)); | |
| } | |
| return array; | |
| } | |
| Result<std::unique_ptr<MappedFields>> MappedFieldsFromJson(const nlohmann::json& json) { | |
| if (!json.is_array()) [[unlikely]] { | |
| return JsonParseError("Cannot parse non-array mapping fields (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| std::vector<MappedField> fields; | |
| for (const auto& field_json : json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto field, MappedFieldFromJson(field_json)); | |
| fields.push_back(std::move(field)); | |
| } | |
| return MappedFields::Make(std::move(fields)); | |
| } | |
| nlohmann::json ToJson(const NameMapping& name_mapping) { | |
| return ToJson(name_mapping.AsMappedFields()); | |
| } | |
| Result<std::unique_ptr<NameMapping>> NameMappingFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto mapped_fields, MappedFieldsFromJson(json)); | |
| return NameMapping::Make(std::move(mapped_fields)); | |
| } | |
| Result<std::string> UpdateMappingFromJsonString( | |
| std::string_view mapping_json, | |
| const std::unordered_map<int32_t, std::shared_ptr<SchemaField>>& updates, | |
| const std::multimap<int32_t, int32_t>& adds) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(std::string(mapping_json))); | |
| ICEBERG_ASSIGN_OR_RAISE(auto current_mapping, NameMappingFromJson(json)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto updated_mapping, | |
| UpdateMapping(*current_mapping, updates, adds)); | |
| return ToJsonString(ToJson(*updated_mapping)); | |
| } | |
| nlohmann::json ToJson(const TableIdentifier& identifier) { | |
| nlohmann::json json; | |
| json[kNamespace] = identifier.ns.levels; | |
| json[kName] = identifier.name; | |
| return json; | |
| } | |
| Result<TableIdentifier> TableIdentifierFromJson(const nlohmann::json& json) { | |
| TableIdentifier identifier; | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| identifier.ns.levels, | |
| GetJsonValueOrDefault<std::vector<std::string>>(json, kNamespace)); | |
| ICEBERG_ASSIGN_OR_RAISE(identifier.name, GetJsonValue<std::string>(json, kName)); | |
| return identifier; | |
| } | |
| nlohmann::json ToJson(const Namespace& ns) { return ns.levels; } | |
| Result<Namespace> NamespaceFromJson(const nlohmann::json& json) { | |
| if (!json.is_array()) [[unlikely]] { | |
| return JsonParseError("Cannot parse namespace from non-array (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| Namespace ns; | |
| ICEBERG_ASSIGN_OR_RAISE(ns.levels, GetTypedJsonValue<std::vector<std::string>>(json)); | |
| return ns; | |
| } | |
| Result<nlohmann::json> ToJson(const TableUpdate& update) { | |
| nlohmann::json json; | |
| switch (update.kind()) { | |
| case TableUpdate::Kind::kAssignUUID: { | |
| const auto& u = internal::checked_cast<const table::AssignUUID&>(update); | |
| json[kAction] = kActionAssignUUID; | |
| json[kUUID] = u.uuid(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kUpgradeFormatVersion: { | |
| const auto& u = internal::checked_cast<const table::UpgradeFormatVersion&>(update); | |
| json[kAction] = kActionUpgradeFormatVersion; | |
| json[kFormatVersion] = u.format_version(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kAddSchema: { | |
| const auto& u = internal::checked_cast<const table::AddSchema&>(update); | |
| json[kAction] = kActionAddSchema; | |
| if (u.schema()) { | |
| ICEBERG_ASSIGN_OR_RAISE(json[kSchema], ToJson(*u.schema())); | |
| } else { | |
| json[kSchema] = nlohmann::json::value_t::null; | |
| } | |
| json[kLastColumnId] = u.last_column_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetCurrentSchema: { | |
| const auto& u = internal::checked_cast<const table::SetCurrentSchema&>(update); | |
| json[kAction] = kActionSetCurrentSchema; | |
| json[kSchemaId] = u.schema_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kAddPartitionSpec: { | |
| const auto& u = internal::checked_cast<const table::AddPartitionSpec&>(update); | |
| json[kAction] = kActionAddPartitionSpec; | |
| if (u.spec()) { | |
| json[kSpec] = ToJson(*u.spec()); | |
| } else { | |
| json[kSpec] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetDefaultPartitionSpec: { | |
| const auto& u = | |
| internal::checked_cast<const table::SetDefaultPartitionSpec&>(update); | |
| json[kAction] = kActionSetDefaultPartitionSpec; | |
| json[kSpecId] = u.spec_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemovePartitionSpecs: { | |
| const auto& u = internal::checked_cast<const table::RemovePartitionSpecs&>(update); | |
| json[kAction] = kActionRemovePartitionSpecs; | |
| json[kSpecIds] = u.spec_ids(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveSchemas: { | |
| const auto& u = internal::checked_cast<const table::RemoveSchemas&>(update); | |
| json[kAction] = kActionRemoveSchemas; | |
| json[kSchemaIds] = u.schema_ids(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kAddSortOrder: { | |
| const auto& u = internal::checked_cast<const table::AddSortOrder&>(update); | |
| json[kAction] = kActionAddSortOrder; | |
| if (u.sort_order()) { | |
| json[kSortOrder] = ToJson(*u.sort_order()); | |
| } else { | |
| json[kSortOrder] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetDefaultSortOrder: { | |
| const auto& u = internal::checked_cast<const table::SetDefaultSortOrder&>(update); | |
| json[kAction] = kActionSetDefaultSortOrder; | |
| json[kSortOrderId] = u.sort_order_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kAddSnapshot: { | |
| const auto& u = internal::checked_cast<const table::AddSnapshot&>(update); | |
| json[kAction] = kActionAddSnapshot; | |
| if (u.snapshot()) { | |
| json[kSnapshot] = ToJson(*u.snapshot()); | |
| } else { | |
| json[kSnapshot] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveSnapshots: { | |
| const auto& u = internal::checked_cast<const table::RemoveSnapshots&>(update); | |
| json[kAction] = kActionRemoveSnapshots; | |
| json[kSnapshotIds] = u.snapshot_ids(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveSnapshotRef: { | |
| const auto& u = internal::checked_cast<const table::RemoveSnapshotRef&>(update); | |
| json[kAction] = kActionRemoveSnapshotRef; | |
| json[kRefName] = u.ref_name(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetSnapshotRef: { | |
| const auto& u = internal::checked_cast<const table::SetSnapshotRef&>(update); | |
| json[kAction] = kActionSetSnapshotRef; | |
| json[kRefName] = u.ref_name(); | |
| json[kSnapshotId] = u.snapshot_id(); | |
| json[kType] = ToString(u.type()); | |
| SetOptionalField(json, kMinSnapshotsToKeep, u.min_snapshots_to_keep()); | |
| SetOptionalField(json, kMaxSnapshotAgeMs, u.max_snapshot_age_ms()); | |
| SetOptionalField(json, kMaxRefAgeMs, u.max_ref_age_ms()); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetProperties: { | |
| const auto& u = internal::checked_cast<const table::SetProperties&>(update); | |
| json[kAction] = kActionSetProperties; | |
| json[kUpdates] = u.updated(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveProperties: { | |
| const auto& u = internal::checked_cast<const table::RemoveProperties&>(update); | |
| json[kAction] = kActionRemoveProperties; | |
| json[kRemovals] = std::vector<std::string>(u.removed().begin(), u.removed().end()); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetLocation: { | |
| const auto& u = internal::checked_cast<const table::SetLocation&>(update); | |
| json[kAction] = kActionSetLocation; | |
| json[kLocation] = u.location(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetStatistics: { | |
| const auto& u = internal::checked_cast<const table::SetStatistics&>(update); | |
| json[kAction] = kActionSetStatistics; | |
| if (u.statistics_file()) { | |
| json[kStatistics] = ToJson(*u.statistics_file()); | |
| } else { | |
| json[kStatistics] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveStatistics: { | |
| const auto& u = internal::checked_cast<const table::RemoveStatistics&>(update); | |
| json[kAction] = kActionRemoveStatistics; | |
| json[kSnapshotId] = u.snapshot_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kSetPartitionStatistics: { | |
| const auto& u = | |
| internal::checked_cast<const table::SetPartitionStatistics&>(update); | |
| json[kAction] = kActionSetPartitionStatistics; | |
| if (u.partition_statistics_file()) { | |
| json[kPartitionStatistics] = ToJson(*u.partition_statistics_file()); | |
| } else { | |
| json[kPartitionStatistics] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemovePartitionStatistics: { | |
| const auto& u = | |
| internal::checked_cast<const table::RemovePartitionStatistics&>(update); | |
| json[kAction] = kActionRemovePartitionStatistics; | |
| json[kSnapshotId] = u.snapshot_id(); | |
| break; | |
| } | |
| case TableUpdate::Kind::kAddEncryptionKey: { | |
| const auto& u = internal::checked_cast<const table::AddEncryptionKey&>(update); | |
| json[kAction] = kActionAddEncryptionKey; | |
| json[kEncryptionKey] = ToJson(u.key()); | |
| break; | |
| } | |
| case TableUpdate::Kind::kRemoveEncryptionKey: { | |
| const auto& u = internal::checked_cast<const table::RemoveEncryptionKey&>(update); | |
| json[kAction] = kActionRemoveEncryptionKey; | |
| json[kKeyId] = u.key_id(); | |
| break; | |
| } | |
| } | |
| return json; | |
| } | |
| nlohmann::json ToJson(const TableRequirement& requirement) { | |
| nlohmann::json json; | |
| switch (requirement.kind()) { | |
| case TableRequirement::Kind::kAssertDoesNotExist: | |
| json[kType] = kRequirementAssertDoesNotExist; | |
| break; | |
| case TableRequirement::Kind::kAssertUUID: { | |
| const auto& r = internal::checked_cast<const table::AssertUUID&>(requirement); | |
| json[kType] = kRequirementAssertUUID; | |
| json[kUUID] = r.uuid(); | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertRefSnapshotID: { | |
| const auto& r = | |
| internal::checked_cast<const table::AssertRefSnapshotID&>(requirement); | |
| json[kType] = kRequirementAssertRefSnapshotID; | |
| json[kRef] = r.ref_name(); | |
| if (r.snapshot_id().has_value()) { | |
| json[kSnapshotId] = r.snapshot_id().value(); | |
| } else { | |
| json[kSnapshotId] = nlohmann::json::value_t::null; | |
| } | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertLastAssignedFieldId: { | |
| const auto& r = | |
| internal::checked_cast<const table::AssertLastAssignedFieldId&>(requirement); | |
| json[kType] = kRequirementAssertLastAssignedFieldId; | |
| json[kLastAssignedFieldId] = r.last_assigned_field_id(); | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertCurrentSchemaID: { | |
| const auto& r = | |
| internal::checked_cast<const table::AssertCurrentSchemaID&>(requirement); | |
| json[kType] = kRequirementAssertCurrentSchemaID; | |
| json[kCurrentSchemaId] = r.schema_id(); | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertLastAssignedPartitionId: { | |
| const auto& r = internal::checked_cast<const table::AssertLastAssignedPartitionId&>( | |
| requirement); | |
| json[kType] = kRequirementAssertLastAssignedPartitionId; | |
| json[kLastAssignedPartitionId] = r.last_assigned_partition_id(); | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertDefaultSpecID: { | |
| const auto& r = | |
| internal::checked_cast<const table::AssertDefaultSpecID&>(requirement); | |
| json[kType] = kRequirementAssertDefaultSpecID; | |
| json[kDefaultSpecId] = r.spec_id(); | |
| break; | |
| } | |
| case TableRequirement::Kind::kAssertDefaultSortOrderID: { | |
| const auto& r = | |
| internal::checked_cast<const table::AssertDefaultSortOrderID&>(requirement); | |
| json[kType] = kRequirementAssertDefaultSortOrderID; | |
| json[kDefaultSortOrderId] = r.sort_order_id(); | |
| break; | |
| } | |
| } | |
| return json; | |
| } | |
| Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto action, GetJsonValue<std::string>(json, kAction)); | |
| if (action == kActionAssignUUID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto uuid, GetJsonValue<std::string>(json, kUUID)); | |
| return std::make_unique<table::AssignUUID>(std::move(uuid)); | |
| } | |
| if (action == kActionUpgradeFormatVersion) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto format_version, | |
| GetJsonValue<int8_t>(json, kFormatVersion)); | |
| return std::make_unique<table::UpgradeFormatVersion>(format_version); | |
| } | |
| if (action == kActionAddSchema) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_json, | |
| GetJsonValue<nlohmann::json>(json, kSchema)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto parsed_schema, SchemaFromJson(schema_json)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto highest_field_id, parsed_schema->HighestFieldId()); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto last_column_id, | |
| GetJsonValueOrDefault<int32_t>(json, kLastColumnId, highest_field_id)); | |
| return std::make_unique<table::AddSchema>(std::move(parsed_schema), last_column_id); | |
| } | |
| if (action == kActionSetCurrentSchema) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValue<int32_t>(json, kSchemaId)); | |
| return std::make_unique<table::SetCurrentSchema>(schema_id); | |
| } | |
| if (action == kActionAddPartitionSpec) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_json, GetJsonValue<nlohmann::json>(json, kSpec)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec, PartitionSpecFromJson(spec_json)); | |
| return std::make_unique<table::AddPartitionSpec>(std::move(spec)); | |
| } | |
| if (action == kActionSetDefaultPartitionSpec) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kSpecId)); | |
| return std::make_unique<table::SetDefaultPartitionSpec>(spec_id); | |
| } | |
| if (action == kActionRemovePartitionSpecs) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_ids, | |
| GetJsonValue<std::vector<int32_t>>(json, kSpecIds)); | |
| return std::make_unique<table::RemovePartitionSpecs>(std::move(spec_ids)); | |
| } | |
| if (action == kActionRemoveSchemas) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_ids_vec, | |
| GetJsonValue<std::vector<int32_t>>(json, kSchemaIds)); | |
| std::unordered_set<int32_t> schema_ids(schema_ids_vec.begin(), schema_ids_vec.end()); | |
| return std::make_unique<table::RemoveSchemas>(std::move(schema_ids)); | |
| } | |
| if (action == kActionAddSortOrder) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_order_json, | |
| GetJsonValue<nlohmann::json>(json, kSortOrder)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_order, SortOrderFromJson(sort_order_json)); | |
| return std::make_unique<table::AddSortOrder>(std::move(sort_order)); | |
| } | |
| if (action == kActionSetDefaultSortOrder) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_order_id, | |
| GetJsonValue<int32_t>(json, kSortOrderId)); | |
| return std::make_unique<table::SetDefaultSortOrder>(sort_order_id); | |
| } | |
| if (action == kActionAddSnapshot) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_json, | |
| GetJsonValue<nlohmann::json>(json, kSnapshot)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot, SnapshotFromJson(snapshot_json)); | |
| return std::make_unique<table::AddSnapshot>(std::move(snapshot)); | |
| } | |
| if (action == kActionRemoveSnapshots) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_ids, | |
| GetJsonValue<std::vector<int64_t>>(json, kSnapshotIds)); | |
| return std::make_unique<table::RemoveSnapshots>(std::move(snapshot_ids)); | |
| } | |
| if (action == kActionRemoveSnapshotRef) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRefName)); | |
| return std::make_unique<table::RemoveSnapshotRef>(std::move(ref_name)); | |
| } | |
| if (action == kActionSetSnapshotRef) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRefName)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto type, | |
| GetJsonValue<std::string>(json, kType).and_then(SnapshotRefTypeFromString)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto min_snapshots, | |
| GetJsonValueOptional<int32_t>(json, kMinSnapshotsToKeep)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto max_snapshot_age, | |
| GetJsonValueOptional<int64_t>(json, kMaxSnapshotAgeMs)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto max_ref_age, | |
| GetJsonValueOptional<int64_t>(json, kMaxRefAgeMs)); | |
| if (type == SnapshotRefType::kTag) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto tag, SnapshotRef::MakeTag(snapshot_id, max_ref_age)); | |
| return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), *tag); | |
| } else { | |
| ICEBERG_CHECK(type == SnapshotRefType::kBranch, | |
| "Expected branch type for snapshot ref"); | |
| ICEBERG_ASSIGN_OR_RAISE(auto branch, | |
| SnapshotRef::MakeBranch(snapshot_id, min_snapshots, | |
| max_snapshot_age, max_ref_age)); | |
| return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), *branch); | |
| } | |
| } | |
| if (action == kActionSetProperties) { | |
| using StringMap = std::unordered_map<std::string, std::string>; | |
| ICEBERG_ASSIGN_OR_RAISE(auto updates, | |
| json.contains(kUpdates) || !json.contains(kUpdated) | |
| ? GetJsonValue<StringMap>(json, kUpdates) | |
| : GetJsonValue<StringMap>(json, kUpdated)); | |
| return std::make_unique<table::SetProperties>(std::move(updates)); | |
| } | |
| if (action == kActionRemoveProperties) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto removals_vec, | |
| json.contains(kRemovals) || !json.contains(kRemoved) | |
| ? GetJsonValue<std::vector<std::string>>(json, kRemovals) | |
| : GetJsonValue<std::vector<std::string>>(json, kRemoved)); | |
| std::unordered_set<std::string> removals( | |
| std::make_move_iterator(removals_vec.begin()), | |
| std::make_move_iterator(removals_vec.end())); | |
| return std::make_unique<table::RemoveProperties>(std::move(removals)); | |
| } | |
| if (action == kActionSetLocation) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto location, GetJsonValue<std::string>(json, kLocation)); | |
| return std::make_unique<table::SetLocation>(std::move(location)); | |
| } | |
| if (action == kActionSetStatistics) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto statistics_json, | |
| GetJsonValue<nlohmann::json>(json, kStatistics)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto statistics_file, | |
| StatisticsFileFromJson(statistics_json)); | |
| return std::make_unique<table::SetStatistics>(std::move(statistics_file)); | |
| } | |
| if (action == kActionRemoveStatistics) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId)); | |
| return std::make_unique<table::RemoveStatistics>(snapshot_id); | |
| } | |
| if (action == kActionSetPartitionStatistics) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_statistics_json, | |
| GetJsonValue<nlohmann::json>(json, kPartitionStatistics)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_statistics_file, | |
| PartitionStatisticsFileFromJson(partition_statistics_json)); | |
| return std::make_unique<table::SetPartitionStatistics>( | |
| std::move(partition_statistics_file)); | |
| } | |
| if (action == kActionRemovePartitionStatistics) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId)); | |
| return std::make_unique<table::RemovePartitionStatistics>(snapshot_id); | |
| } | |
| if (action == kActionAddEncryptionKey) { | |
| if (!json.contains(kEncryptionKey)) { | |
| return JsonParseError("Invalid encryption key, must be non-null object: null"); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto key, EncryptedKeyFromJson(json.at(kEncryptionKey))); | |
| return std::make_unique<table::AddEncryptionKey>(std::move(key)); | |
| } | |
| if (action == kActionRemoveEncryptionKey) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValue<std::string>(json, kKeyId)); | |
| return std::make_unique<table::RemoveEncryptionKey>(std::move(key_id)); | |
| } | |
| return JsonParseError("Unknown table update action: {}", action); | |
| } | |
| Result<std::unique_ptr<TableRequirement>> TableRequirementFromJson( | |
| const nlohmann::json& json) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto type, GetJsonValue<std::string>(json, kType)); | |
| if (type == kRequirementAssertDoesNotExist) { | |
| return std::make_unique<table::AssertDoesNotExist>(); | |
| } | |
| if (type == kRequirementAssertUUID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto uuid, GetJsonValue<std::string>(json, kUUID)); | |
| return std::make_unique<table::AssertUUID>(std::move(uuid)); | |
| } | |
| if (type == kRequirementAssertRefSnapshotID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRef)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id_opt, | |
| GetJsonValueOptional<int64_t>(json, kSnapshotId)); | |
| return std::make_unique<table::AssertRefSnapshotID>(std::move(ref_name), | |
| snapshot_id_opt); | |
| } | |
| if (type == kRequirementAssertLastAssignedFieldId) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto last_assigned_field_id, | |
| GetJsonValue<int32_t>(json, kLastAssignedFieldId)); | |
| return std::make_unique<table::AssertLastAssignedFieldId>(last_assigned_field_id); | |
| } | |
| if (type == kRequirementAssertCurrentSchemaID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_id, | |
| GetJsonValue<int32_t>(json, kCurrentSchemaId)); | |
| return std::make_unique<table::AssertCurrentSchemaID>(schema_id); | |
| } | |
| if (type == kRequirementAssertLastAssignedPartitionId) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto last_assigned_partition_id, | |
| GetJsonValue<int32_t>(json, kLastAssignedPartitionId)); | |
| return std::make_unique<table::AssertLastAssignedPartitionId>( | |
| last_assigned_partition_id); | |
| } | |
| if (type == kRequirementAssertDefaultSpecID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kDefaultSpecId)); | |
| return std::make_unique<table::AssertDefaultSpecID>(spec_id); | |
| } | |
| if (type == kRequirementAssertDefaultSortOrderID) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto sort_order_id, | |
| GetJsonValue<int32_t>(json, kDefaultSortOrderId)); | |
| return std::make_unique<table::AssertDefaultSortOrderID>(sort_order_id); | |
| } | |
| return JsonParseError("Unknown table requirement type: {}", type); | |
| } | |
| Result<DataFile> DataFileFromJson( | |
| const nlohmann::json& json, | |
| const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& | |
| partition_spec_by_id, | |
| const Schema& schema) { | |
| if (!json.is_object()) { | |
| return JsonParseError("DataFile must be a JSON object (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| DataFile data_file; | |
| ICEBERG_ASSIGN_OR_RAISE(auto content_str, GetJsonValue<std::string>(json, kContent)); | |
| if (content_str == kContentData || content_str == "DATA") { | |
| data_file.content = DataFile::Content::kData; | |
| } else if (content_str == kContentPositionDeletes || | |
| content_str == "POSITION_DELETES") { | |
| data_file.content = DataFile::Content::kPositionDeletes; | |
| } else if (content_str == kContentEqualityDeletes || | |
| content_str == "EQUALITY_DELETES") { | |
| data_file.content = DataFile::Content::kEqualityDeletes; | |
| } else { | |
| return JsonParseError("Unknown data file content: {}", content_str); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.file_path, | |
| GetJsonValue<std::string>(json, kFilePath)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto format_str, GetJsonValue<std::string>(json, kFileFormat)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.file_format, FileFormatTypeFromString(format_str)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonInteger<int32_t>(json, kSpecId)); | |
| data_file.partition_spec_id = spec_id; | |
| auto it = partition_spec_by_id.find(spec_id); | |
| if (it == partition_spec_by_id.end() || !it->second) { | |
| return JsonParseError("Invalid partition spec id: {}", spec_id); | |
| } | |
| if (json.contains(kPartition)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_vals, | |
| GetJsonValue<nlohmann::json>(json, kPartition)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto struct_type, it->second->PartitionType(schema)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.partition, | |
| PartitionValuesFromJson(partition_vals, *struct_type)); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.record_count, | |
| GetJsonInteger<int64_t>(json, kRecordCount)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.file_size_in_bytes, | |
| GetJsonInteger<int64_t>(json, kFileSizeInBytes)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.column_sizes, | |
| KeyValueMapFromJson<int64_t>(json, kColumnSizes)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.value_counts, | |
| KeyValueMapFromJson<int64_t>(json, kValueCounts)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.null_value_counts, | |
| KeyValueMapFromJson<int64_t>(json, kNullValueCounts)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.nan_value_counts, | |
| KeyValueMapFromJson<int64_t>(json, kNanValueCounts)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.lower_bounds, BytesMapFromJson(json, kLowerBounds)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.upper_bounds, BytesMapFromJson(json, kUpperBounds)); | |
| if (json.contains(kKeyMetadata) && !json.at(kKeyMetadata).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto key_metadata, | |
| GetJsonValue<std::string>(json, kKeyMetadata)); | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.key_metadata, | |
| StringUtils::HexStringToBytes(key_metadata)); | |
| } | |
| if (json.contains(kSplitOffsets) && !json.at(kSplitOffsets).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.split_offsets, | |
| IntegerVectorFromJson<int64_t>(json, kSplitOffsets)); | |
| } | |
| if (json.contains(kEqualityIds) && !json.at(kEqualityIds).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.equality_ids, | |
| IntegerVectorFromJson<int32_t>(json, kEqualityIds)); | |
| } | |
| if (json.contains(kSortOrderId) && !json.at(kSortOrderId).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.sort_order_id, | |
| GetJsonInteger<int32_t>(json, kSortOrderId)); | |
| } | |
| if (json.contains(kFirstRowId) && !json.at(kFirstRowId).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.first_row_id, | |
| GetJsonInteger<int64_t>(json, kFirstRowId)); | |
| } | |
| if (json.contains(kReferencedDataFile) && !json.at(kReferencedDataFile).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.referenced_data_file, | |
| GetJsonValue<std::string>(json, kReferencedDataFile)); | |
| } | |
| if (json.contains(kContentOffset) && !json.at(kContentOffset).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.content_offset, | |
| GetJsonInteger<int64_t>(json, kContentOffset)); | |
| } | |
| if (json.contains(kContentSizeInBytes) && !json.at(kContentSizeInBytes).is_null()) { | |
| ICEBERG_ASSIGN_OR_RAISE(data_file.content_size_in_bytes, | |
| GetJsonInteger<int64_t>(json, kContentSizeInBytes)); | |
| } | |
| return data_file; | |
| } | |
| Result<nlohmann::json> ToJson( | |
| const DataFile& data_file, | |
| const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& | |
| partition_specs_by_id, | |
| const Schema& schema) { | |
| if (!data_file.partition_spec_id.has_value()) { | |
| return ValidationFailed("Invalid partition spec id from content file: null"); | |
| } | |
| auto it = partition_specs_by_id.find(data_file.partition_spec_id.value()); | |
| if (it == partition_specs_by_id.end() || !it->second) { | |
| return ValidationFailed("Invalid partition spec: null"); | |
| } | |
| if (data_file.partition_spec_id.value() != it->second->spec_id()) { | |
| return ValidationFailed( | |
| "Invalid partition spec id from content file: expected = {}, actual = {}", | |
| it->second->spec_id(), data_file.partition_spec_id.value()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto partition_type, it->second->PartitionType(schema)); | |
| if (data_file.partition.num_fields() != partition_type->fields().size()) { | |
| return ValidationFailed( | |
| "Invalid partition data from content file: expected = {}, actual = {}", | |
| partition_type->fields().empty() ? "unpartitioned" : "partitioned", | |
| data_file.partition.num_fields() == 0 ? "unpartitioned" : "partitioned"); | |
| } | |
| for (size_t pos = 0; pos < partition_type->fields().size(); ++pos) { | |
| const auto& literal = data_file.partition.values()[pos]; | |
| const auto& expected_type = partition_type->fields()[pos].type(); | |
| if (!literal.IsNull() && (!literal.type() || *literal.type() != *expected_type)) { | |
| return ValidationFailed( | |
| "Invalid partition value type at position {}: expected = {}, actual = {}", pos, | |
| expected_type->ToString(), | |
| literal.type() ? literal.type()->ToString() : "unknown"); | |
| } | |
| } | |
| return DataFileToJsonUnchecked(data_file); | |
| } | |
| Result<nlohmann::json> ToJson( | |
| const FileScanTask& task, | |
| const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& | |
| partition_specs_by_id, | |
| const Schema& schema) { | |
| if (!task.data_file()) { | |
| return ValidationFailed("Cannot serialize FileScanTask without data-file"); | |
| } | |
| if (task.data_file()->content != DataFile::Content::kData) { | |
| return ValidationFailed("FileScanTask data-file must have data content"); | |
| } | |
| // The shared data file may have changed size; don't write an invalid range. | |
| auto range_result = | |
| FileScanTask::MakeSplit(task.data_file(), task.start(), task.length()); | |
| if (!range_result) { | |
| return ValidationFailed("Invalid FileScanTask range: {}", | |
| range_result.error().message); | |
| } | |
| nlohmann::json json; | |
| json[kTaskType] = kFileScanTaskType; | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_json, ToJson(schema)); | |
| json[kSchema] = std::move(schema_json); | |
| ICEBERG_ASSIGN_OR_RAISE(auto data_file_json, | |
| ToJson(*task.data_file(), partition_specs_by_id, schema)); | |
| const auto spec_id = task.data_file()->partition_spec_id.value(); | |
| const auto spec_it = partition_specs_by_id.find(spec_id); | |
| if (spec_it == partition_specs_by_id.end() || !spec_it->second) { | |
| return ValidationFailed("Invalid partition spec: null"); | |
| } | |
| json[kSpec] = ToJson(*spec_it->second); | |
| json[kDataFile] = std::move(data_file_json); | |
| json[kStart] = task.start(); | |
| json[kLength] = task.length(); | |
| nlohmann::json delete_files_json = nlohmann::json::array(); | |
| for (const auto& delete_file : task.delete_files()) { | |
| if (!delete_file) { | |
| return ValidationFailed("FileScanTask delete-files must not contain null"); | |
| } | |
| if (delete_file->content == DataFile::Content::kData) { | |
| return ValidationFailed("FileScanTask delete-file must have delete content"); | |
| } | |
| if (delete_file->partition_spec_id != task.data_file()->partition_spec_id) { | |
| return ValidationFailed( | |
| "Invalid partition spec id from content file: expected = {}, actual = {}", | |
| spec_id, | |
| delete_file->partition_spec_id.has_value() | |
| ? std::to_string(delete_file->partition_spec_id.value()) | |
| : "null"); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto delete_file_json, | |
| ToJson(*delete_file, partition_specs_by_id, schema)); | |
| delete_files_json.push_back(std::move(delete_file_json)); | |
| } | |
| if (!delete_files_json.empty()) { | |
| json[kDeleteFiles] = std::move(delete_files_json); | |
| } | |
| if (task.residual_filter()) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto residual_json, ToJson(*task.residual_filter())); | |
| json[kResidualFilter] = std::move(residual_json); | |
| } | |
| return json; | |
| } | |
| Result<std::shared_ptr<FileScanTask>> FileScanTaskFromJson(const nlohmann::json& json) { | |
| if (!json.is_object()) { | |
| return JsonParseError( | |
| "Cannot parse file scan task from a non-object (JSON value of type {})", | |
| json.type_name()); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto task_type, | |
| GetJsonValueOptional<std::string>(json, kTaskType)); | |
| if (task_type.has_value() && | |
| !StringUtils::EqualsIgnoreCase(*task_type, kFileScanTaskType)) { | |
| return JsonParseError("Unsupported scan task type: {}", *task_type); | |
| } | |
| if (json.contains(kDeleteFileReferences) && !json.at(kDeleteFileReferences).is_null()) { | |
| return JsonParseError( | |
| "Cannot parse FileScanTask with 'delete-file-references'; use " | |
| "rest::FileScanTasksFromJson for REST scan responses"); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto schema, SchemaFromJson(schema_json)); | |
| auto shared_schema = std::shared_ptr<Schema>(std::move(schema)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_json, GetJsonValue<nlohmann::json>(json, kSpec)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonInteger<int32_t>(spec_json, kSpecId)); | |
| // An embedded task spec may reference columns dropped from its embedded schema. | |
| ICEBERG_ASSIGN_OR_RAISE(auto spec, | |
| ParseBoundPartitionSpec(*shared_schema, spec_json, spec_id, | |
| /*allow_missing_fields=*/true)); | |
| auto shared_spec = std::shared_ptr<PartitionSpec>(std::move(spec)); | |
| const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> partition_spec_by_id{ | |
| {shared_spec->spec_id(), shared_spec}}; | |
| ICEBERG_ASSIGN_OR_RAISE(auto data_file_json, | |
| GetJsonValue<nlohmann::json>(json, kDataFile)); | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto data_file, | |
| DataFileFromJson(data_file_json, partition_spec_by_id, *shared_schema)); | |
| if (data_file.content != DataFile::Content::kData) { | |
| return JsonParseError("FileScanTask data-file must have data content"); | |
| } | |
| ICEBERG_ASSIGN_OR_RAISE(auto start, GetJsonInteger<int64_t>(json, kStart)); | |
| ICEBERG_ASSIGN_OR_RAISE(auto length, GetJsonInteger<int64_t>(json, kLength)); | |
| std::vector<std::shared_ptr<DataFile>> delete_files; | |
| if (json.contains(kDeleteFiles)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto delete_files_json, | |
| GetJsonValue<nlohmann::json>(json, kDeleteFiles)); | |
| if (!delete_files_json.is_array()) { | |
| return JsonParseError( | |
| "Cannot parse delete files from non-array (JSON value of type {})", | |
| delete_files_json.type_name()); | |
| } | |
| for (const auto& delete_file_json : delete_files_json) { | |
| ICEBERG_ASSIGN_OR_RAISE( | |
| auto delete_file, | |
| DataFileFromJson(delete_file_json, partition_spec_by_id, *shared_schema)); | |
| if (delete_file.content == DataFile::Content::kData) { | |
| return JsonParseError("FileScanTask delete-file must have delete content"); | |
| } | |
| delete_files.push_back(std::make_shared<DataFile>(std::move(delete_file))); | |
| } | |
| } | |
| std::shared_ptr<Expression> residual_filter = Expressions::AlwaysTrue(); | |
| if (json.contains(kResidualFilter)) { | |
| ICEBERG_ASSIGN_OR_RAISE(auto filter_json, | |
| GetJsonValue<nlohmann::json>(json, kResidualFilter)); | |
| ICEBERG_ASSIGN_OR_RAISE(residual_filter, ExpressionFromJson(filter_json)); | |
| } | |
| auto task = FileScanTask::MakeSplit(std::make_shared<DataFile>(std::move(data_file)), | |
| start, length, std::move(delete_files), | |
| std::move(residual_filter)); | |
| if (!task) { | |
| return JsonParseError("Invalid FileScanTask range: {}", task.error().message); | |
| } | |
| return std::move(task.value()); | |
| } | |
| } // namespace iceberg |