diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index eee720613..978bbf4c6 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -84,6 +84,7 @@ set(ICEBERG_SOURCES manifest/v1_metadata.cc manifest/v2_metadata.cc manifest/v3_metadata.cc + manifest/v4_types.cc metadata_columns.cc metrics_config.cc metrics/commit_report.cc diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h index 6bcc5b742..239c3970b 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/src/iceberg/cpp20_compatibility_internal.h @@ -102,6 +102,7 @@ #include "iceberg/manifest/manifest_reader.h" #include "iceberg/manifest/manifest_writer.h" #include "iceberg/manifest/rolling_manifest_writer.h" +#include "iceberg/manifest/v4_types.h" #include "iceberg/metadata_columns.h" #include "iceberg/metrics.h" #include "iceberg/metrics/commit_report.h" diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 7eb204b43..bf9d7020f 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -38,7 +38,7 @@ namespace iceberg { namespace { -constexpr std::array, 6> kSnapshotColumns = { +constexpr std::array, 7> kSnapshotColumns = { {{.field_id = 1, .append = [](ArrowArray* column, const Snapshot& row) { @@ -65,13 +65,19 @@ constexpr std::array, 6> kSnapshotColumn return row.manifest_list.empty() ? AppendNull(column) : AppendString(column, row.manifest_list); }}, - {.field_id = 6, .append = [](ArrowArray* column, const Snapshot& row) { - if (row.summary.empty()) { - return AppendNull(column); - } - auto summary = row.summary; - summary.erase(SnapshotSummaryFields::kOperation); - return AppendStringMap(column, summary); + {.field_id = 6, + .append = + [](ArrowArray* column, const Snapshot& row) { + if (row.summary.empty()) { + return AppendNull(column); + } + auto summary = row.summary; + summary.erase(SnapshotSummaryFields::kOperation); + return AppendStringMap(column, summary); + }}, + {.field_id = 9, .append = [](ArrowArray* column, const Snapshot& row) { + return row.root_manifest ? AppendString(column, *row.root_manifest) + : AppendNull(column); }}}}; } // namespace @@ -96,7 +102,8 @@ const std::shared_ptr& SnapshotsTable::CanonicalSchema() { SchemaField::MakeOptional( 6, "summary", std::make_shared(SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); + SchemaField::MakeRequired(8, "value", string()))), + SchemaField::MakeOptional(9, "root_manifest", string())}); return schema; } diff --git a/src/iceberg/json_serde.cc b/src/iceberg/json_serde.cc index 37f6c08c9..48b3014e1 100644 --- a/src/iceberg/json_serde.cc +++ b/src/iceberg/json_serde.cc @@ -114,6 +114,7 @@ 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 kRootManifest = "root-manifest"; constexpr std::string_view kSummary = "summary"; constexpr std::string_view kFirstRowId = "first-row-id"; constexpr std::string_view kAddedRows = "added-rows"; @@ -739,7 +740,11 @@ nlohmann::json ToJson(const Snapshot& snapshot) { json[kSequenceNumber] = snapshot.sequence_number; } json[kTimestampMs] = UnixMsFromTimePointMs(snapshot.timestamp_ms); - json[kManifestList] = snapshot.manifest_list; + if (snapshot.root_manifest.has_value()) { + json[kRootManifest] = snapshot.root_manifest.value(); + } else { + json[kManifestList] = snapshot.manifest_list; + } // If there is an operation, write the summary map if (snapshot.Operation().has_value()) { json[kSummary] = snapshot.summary; @@ -1123,14 +1128,31 @@ Result> SnapshotRefFromJson(const nlohmann::json& j } } -Result> SnapshotFromJson(const nlohmann::json& json) { +Result> SnapshotFromJson(const nlohmann::json& json, + int8_t format_version) { + if (format_version < 1 || + format_version > TableMetadata::kSupportedTableFormatVersion) { + return JsonParseError("Unsupported snapshot format version: {}", format_version); + } ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue(json, kSnapshotId)); ICEBERG_ASSIGN_OR_RAISE(auto sequence_number, GetJsonValueOptional(json, kSequenceNumber)); ICEBERG_ASSIGN_OR_RAISE(auto unix_ms, GetJsonValue(json, kTimestampMs)); auto timestamp_ms = TimePointMsFromUnixMs(unix_ms); ICEBERG_ASSIGN_OR_RAISE(auto manifest_list, - GetJsonValue(json, kManifestList)); + GetJsonValueOptional(json, kManifestList)); + ICEBERG_ASSIGN_OR_RAISE(auto root_manifest, + GetJsonValueOptional(json, kRootManifest)); + if (manifest_list.has_value() == root_manifest.has_value()) { + return JsonParseError( + "Snapshot requires exactly one of manifest-list or root-manifest"); + } + ICEBERG_ASSIGN_OR_RAISE( + auto constraint_statuses, + GetJsonValueOptional(json, "constraint-statuses")); + if (constraint_statuses.has_value() && !constraint_statuses->empty()) { + return NotSupported("V4 snapshot constraint statuses are not supported yet"); + } ICEBERG_ASSIGN_OR_RAISE(auto parent_snapshot_id, GetJsonValueOptional(json, kParentSnapshotId)); @@ -1148,14 +1170,15 @@ Result> SnapshotFromJson(const nlohmann::json& json) { "Invalid snapshot summary field value for '{}' (JSON value of type {})", key, value.type_name()); } - if (key == SnapshotSummaryFields::kOperation && + if (key == SnapshotSummaryFields::kOperation && !root_manifest.has_value() && !kValidDataOperation.contains(value.get())) { return JsonParseError("Invalid snapshot operation in snapshot summary"); } summary[key] = value.get(); } // If summary is available but operation is missing, set operation to overwrite. - if (!summary.contains(SnapshotSummaryFields::kOperation)) { + if (!root_manifest.has_value() && + !summary.contains(SnapshotSummaryFields::kOperation)) { summary[SnapshotSummaryFields::kOperation] = DataOperation::kOverwrite; } } @@ -1165,29 +1188,36 @@ Result> SnapshotFromJson(const nlohmann::json& json) { ICEBERG_ASSIGN_OR_RAISE(auto added_rows, GetJsonValueOptional(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; + if (!root_manifest.has_value()) { + 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(json, kSchemaId)); ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValueOptional(json, kKeyId)); - return std::make_unique( + auto snapshot = std::make_unique( 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)); + manifest_list.value_or(""), std::move(summary), schema_id, first_row_id, added_rows, + std::move(key_id), std::move(root_manifest)); + if (auto status = snapshot->ValidateForFormatVersion(format_version); !status) { + return JsonParseError("Invalid snapshot: {}", status.error().message); + } + return snapshot; } nlohmann::json ToJson(const BlobMetadata& blob_metadata) { @@ -1345,11 +1375,26 @@ Result EncryptedKeyFromJson(const nlohmann::json& json) { } Result ToJson(const TableMetadata& table_metadata) { + ICEBERG_PRECHECK( + table_metadata.format_version >= 1 && + table_metadata.format_version <= TableMetadata::kSupportedTableFormatVersion, + "Unsupported table format version: {}", table_metadata.format_version); + for (const auto& snapshot : table_metadata.snapshots) { + ICEBERG_PRECHECK(snapshot != nullptr, "Cannot write null snapshot"); + if (auto status = snapshot->ValidateForFormatVersion(table_metadata.format_version); + !status) { + return InvalidArgument("Cannot write snapshot: {}", status.error().message); + } + } nlohmann::json json; json[kFormatVersion] = table_metadata.format_version; json[kTableUuid] = table_metadata.table_uuid; - json[kLocation] = table_metadata.location; + ICEBERG_RETURN_UNEXPECTED(table_metadata.ValidateLocation()); + if (table_metadata.format_version < TableMetadata::kMinFormatVersionOptionalLocation || + !table_metadata.location.empty()) { + json[kLocation] = table_metadata.location; + } if (table_metadata.format_version > 1) { json[kLastSequenceNumber] = table_metadata.last_sequence_number; } @@ -1374,6 +1419,12 @@ Result ToJson(const TableMetadata& table_metadata) { // 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) { + if (table_metadata.format_version >= TableMetadata::kMinFormatVersionConstraints && + !schema->IdentifierFieldIds().empty()) { + return NotSupported( + "V4 primary-key constraints are not supported yet; " + "identifier-field-ids cannot be written in v4 schemas"); + } ICEBERG_ASSIGN_OR_RAISE(auto schema_json, ToJson(*schema)); schemas_json.push_back(std::move(schema_json)); } @@ -1455,9 +1506,19 @@ Result> ParseSchemas( ICEBERG_ASSIGN_OR_RAISE(current_schema_id, GetJsonInteger(json, kCurrentSchemaId)); + const auto parse_schema = + [format_version]( + const nlohmann::json& schema_json) -> Result> { + if (format_version >= TableMetadata::kMinFormatVersionConstraints && + schema_json.is_object() && schema_json.contains(kIdentifierFieldIds)) { + auto v4_schema_json = schema_json; + v4_schema_json.erase(kIdentifierFieldIds); + return SchemaFromJson(v4_schema_json); + } + return SchemaFromJson(schema_json); + }; for (const auto& schema_json : schema_array) { - ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr schema, - SchemaFromJson(schema_json)); + ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr schema, parse_schema(schema_json)); ICEBERG_RETURN_UNEXPECTED(schema->Validate(format_version)); if (schema->schema_id() == current_schema_id) { current_schema = schema; @@ -1594,18 +1655,40 @@ Result> TableMetadataFromJson(const nlohmann::jso auto table_metadata = std::make_unique(); - ICEBERG_ASSIGN_OR_RAISE(table_metadata->format_version, - GetJsonValue(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(auto format_version, + GetJsonInteger(json, kFormatVersion)); + if (format_version < 1 || + format_version > TableMetadata::kSupportedTableFormatVersion) { + return JsonParseError("Cannot read unsupported version: {}", format_version); } + table_metadata->format_version = static_cast(format_version); ICEBERG_ASSIGN_OR_RAISE(table_metadata->table_uuid, GetJsonValueOrDefault(json, kTableUuid)); - ICEBERG_ASSIGN_OR_RAISE(table_metadata->location, - GetJsonValue(json, kLocation)); + if (table_metadata->format_version >= TableMetadata::kMinFormatVersionConstraints) { + ICEBERG_ASSIGN_OR_RAISE( + auto constraints, + GetJsonValueOptional(json, "constraints")); + if (constraints.has_value() && !constraints->empty()) { + return NotSupported("V4 table constraints are not supported yet"); + } + ICEBERG_ASSIGN_OR_RAISE(auto last_constraint_id, + GetJsonIntegerOptional(json, "last-constraint-id")); + if (last_constraint_id.value_or(0) != 0) { + return NotSupported("V4 table constraint IDs are not supported yet"); + } + } + if (table_metadata->format_version >= + TableMetadata::kMinFormatVersionOptionalLocation) { + ICEBERG_ASSIGN_OR_RAISE(table_metadata->location, + GetJsonValueOrDefault(json, kLocation)); + } else { + ICEBERG_ASSIGN_OR_RAISE(table_metadata->location, + GetJsonValue(json, kLocation)); + } + if (auto status = table_metadata->ValidateLocation(); !status) { + return JsonParseError("{}", status.error().message); + } if (table_metadata->format_version > 1) { ICEBERG_ASSIGN_OR_RAISE(table_metadata->last_sequence_number, @@ -1682,8 +1765,13 @@ Result> TableMetadataFromJson(const nlohmann::jso }); } - ICEBERG_ASSIGN_OR_RAISE(table_metadata->snapshots, - FromJsonList(json, kSnapshots, SnapshotFromJson)); + ICEBERG_ASSIGN_OR_RAISE( + table_metadata->snapshots, + FromJsonList(json, kSnapshots, + [format_version = table_metadata->format_version]( + const nlohmann::json& snapshot_json) { + return SnapshotFromJson(snapshot_json, format_version); + })); ICEBERG_ASSIGN_OR_RAISE( table_metadata->statistics, FromJsonList(json, kStatistics, StatisticsFileFromJson)); @@ -2082,7 +2170,7 @@ Result> TableUpdateFromJson(const nlohmann::json& j } if (action == kActionUpgradeFormatVersion) { ICEBERG_ASSIGN_OR_RAISE(auto format_version, - GetJsonValue(json, kFormatVersion)); + GetJsonInteger(json, kFormatVersion)); return std::make_unique(format_version); } if (action == kActionAddSchema) { diff --git a/src/iceberg/json_serde_internal.h b/src/iceberg/json_serde_internal.h index 974488328..bb4f36120 100644 --- a/src/iceberg/json_serde_internal.h +++ b/src/iceberg/json_serde_internal.h @@ -247,9 +247,12 @@ ICEBERG_EXPORT nlohmann::json ToJson(const Snapshot& snapshot); /// \brief Deserializes a JSON object into a `Snapshot` object. /// /// \param json The JSON representation of the snapshot. +/// \param format_version The table format, or the latest supported version when +/// parsing a standalone snapshot without table context. /// \return A `Snapshot` object or an error if the conversion fails. ICEBERG_EXPORT Result> SnapshotFromJson( - const nlohmann::json& json); + const nlohmann::json& json, + int8_t format_version = TableMetadata::kSupportedTableFormatVersion); /// \brief Serializes a `StatisticsFile` object to JSON. /// diff --git a/src/iceberg/location_provider.cc b/src/iceberg/location_provider.cc index 70274edc6..c8b317cc9 100644 --- a/src/iceberg/location_provider.cc +++ b/src/iceberg/location_provider.cc @@ -191,11 +191,24 @@ Result ObjectStoreLocationProvider::NewDataLocation( Result> LocationProvider::Make( std::string_view location, const TableProperties& properties) { + const bool object_store = properties.Get(TableProperties::kObjectStoreEnabled); + // Validate once before either provider can derive a data path or table context. + // An explicit absolute URI is sufficient only for the default provider; object + // stores also need the table location to derive their context. + if (location.empty()) { + ICEBERG_PRECHECK(!object_store, + "Object-store location provider requires a table location"); + ICEBERG_PRECHECK( + !LocationUtil::ParseScheme(properties.Get(TableProperties::kWriteDataLocation)) + .empty(), + "Default location provider requires a table location or an absolute URI in '{}'", + TableProperties::kWriteDataLocation.key()); + } location = LocationUtil::StripTrailingSlash(location); // TODO(xxx): create location provider specified by "write.location-provider.impl" - if (properties.Get(TableProperties::kObjectStoreEnabled)) { + if (object_store) { return std::make_unique(location, properties); } else { return std::make_unique(location, properties); diff --git a/src/iceberg/manifest/v4_types.cc b/src/iceberg/manifest/v4_types.cc new file mode 100644 index 000000000..e5802754c --- /dev/null +++ b/src/iceberg/manifest/v4_types.cc @@ -0,0 +1,76 @@ +/* + * 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 "iceberg/manifest/v4_types.h" + +#include "iceberg/schema_field.h" +#include "iceberg/type.h" + +namespace iceberg { + +Result FileTrackingStatusFromInt(int32_t status) { + if (status < static_cast(FileTrackingStatus::kExisting) || + status > static_cast(FileTrackingStatus::kModified)) { + return InvalidArgument("Invalid file tracking status: {}", status); + } + return static_cast(status); +} + +std::shared_ptr FileTracking::Type() { + static const auto kInstance = std::make_shared(std::vector{ + SchemaField::MakeRequired(0, "status", int32()), + SchemaField::MakeOptional(1, "snapshot_id", int64()), + SchemaField::MakeOptional(5, "modified_snapshot_id", int64()), + SchemaField::MakeOptional(3, "sequence_number", int64()), + SchemaField::MakeOptional(4, "file_sequence_number", int64()), + SchemaField::MakeOptional(142, "first_row_id", int64()), + SchemaField::MakeOptional(6, "deleted_positions", binary()), + SchemaField::MakeOptional(7, "replaced_positions", binary())}); + return kInstance; +} + +std::shared_ptr DeletionVectorMetadata::Type() { + static const auto kInstance = std::make_shared( + std::vector{SchemaField::MakeRequired(155, "location", string()), + SchemaField::MakeRequired(144, "offset", int64()), + SchemaField::MakeRequired(145, "size_in_bytes", int64()), + SchemaField::MakeRequired(156, "cardinality", int64()), + SchemaField::MakeOptional(149, "key_metadata", binary())}); + return kInstance; +} + +std::shared_ptr ManifestInfo::Type() { + static const auto kInstance = std::make_shared(std::vector{ + SchemaField::MakeRequired(521, "format_version", int32()), + SchemaField::MakeOptional(522, "dv", binary()), + SchemaField::MakeRequired(504, "added_files_count", int32()), + SchemaField::MakeRequired(505, "existing_files_count", int32()), + SchemaField::MakeRequired(525, "modified_files_count", int32()), + SchemaField::MakeRequired(506, "deleted_files_count", int32()), + SchemaField::MakeRequired(523, "replaced_files_count", int32()), + SchemaField::MakeRequired(512, "added_rows_count", int64()), + SchemaField::MakeRequired(513, "existing_rows_count", int64()), + SchemaField::MakeRequired(526, "modified_rows_count", int64()), + SchemaField::MakeRequired(514, "deleted_rows_count", int64()), + SchemaField::MakeRequired(524, "replaced_rows_count", int64()), + SchemaField::MakeRequired(516, "min_sequence_number", int64())}); + return kInstance; +} + +} // namespace iceberg diff --git a/src/iceberg/manifest/v4_types.h b/src/iceberg/manifest/v4_types.h new file mode 100644 index 000000000..b44e5d14a --- /dev/null +++ b/src/iceberg/manifest/v4_types.h @@ -0,0 +1,132 @@ +/* + * 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. + */ + +#pragma once + +/// \file iceberg/manifest/v4_types.h +/// Foundational metadata types for the v4 manifest tree. These types do not enable +/// v4 table reads or writes. The v4 specification is still under development. +/// See https://iceberg.apache.org/spec/#manifest-schema. + +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Status of a tracked file in a v4 manifest. +/// Kept separate from ManifestStatus because v1-v3 manifests cannot contain +/// REPLACED or MODIFIED entries. +enum class FileTrackingStatus { + kExisting = 0, + kAdded = 1, + kDeleted = 2, + kReplaced = 3, + kModified = 4, +}; + +ICEBERG_EXPORT Result FileTrackingStatusFromInt(int32_t status); + +/// \brief The v4 tracking struct, shared by data files and manifests. +struct ICEBERG_EXPORT FileTracking { + /// Field id: 0. ADDED, EXISTING, and MODIFIED entries are live. + FileTrackingStatus status = FileTrackingStatus::kAdded; + /// Field id: 1. Snapshot where the file was added, replaced, or deleted. + /// Inherited when null; writers should persist a non-null value. + std::optional snapshot_id; + /// Field id: 5. Snapshot where the file was last modified. + std::optional modified_snapshot_id; + /// Field id: 3. Data sequence number; inherited when null. + std::optional sequence_number; + /// Field id: 4. File sequence number; inherited when null. + std::optional file_sequence_number; + /// Field id: 142. Base row ID; subject to first row ID inheritance. + std::optional first_row_id; + /// Field id: 6. Manifest-only bitmap of positions deleted in modified_snapshot_id. + std::optional> deleted_positions; + /// Field id: 7. Manifest-only bitmap of positions replaced in modified_snapshot_id. + std::optional> replaced_positions; + + constexpr bool IsAlive() const { + return status == FileTrackingStatus::kExisting || + status == FileTrackingStatus::kAdded || + status == FileTrackingStatus::kModified; + } + + static std::shared_ptr Type(); +}; + +/// \brief The v4 deletion_vector struct, colocated with a tracked data file. +/// This references a Puffin blob; it does not hold the bitmap itself. +struct ICEBERG_EXPORT DeletionVectorMetadata { + /// Field id: 155. Location of the file containing the deletion vector. + std::string location; + /// Field id: 144. Must match the blob offset in the Puffin footer. + int64_t offset = 0; + /// Field id: 145. Must match the blob length in the Puffin footer. + int64_t size_in_bytes = 0; + /// Field id: 156. Number of deleted rows (set bits). + int64_t cardinality = 0; + /// Field id: 149. Encryption key metadata, absent for unencrypted blobs. + std::optional> key_metadata; + + static std::shared_ptr Type(); +}; + +/// \brief The v4 manifest_info struct, present only for tracked manifests. +/// File counts include all five tracking statuses and sum to the manifest's +/// record_count. This data carrier does not validate tracked-file requirements. +struct ICEBERG_EXPORT ManifestInfo { + /// Field id: 521. 0 denotes a pre-v4 manifest; 4 denotes a v4 manifest. + int32_t format_version = 0; + /// Field id: 522. Bitmap of positions that are no longer live in a leaf manifest. + std::optional> dv; + /// Field id: 504. + int32_t added_files_count = 0; + /// Field id: 505. + int32_t existing_files_count = 0; + /// Field id: 525. + int32_t modified_files_count = 0; + /// Field id: 506. + int32_t deleted_files_count = 0; + /// Field id: 523. + int32_t replaced_files_count = 0; + /// Field id: 512. + int64_t added_rows_count = 0; + /// Field id: 513. + int64_t existing_rows_count = 0; + /// Field id: 526. + int64_t modified_rows_count = 0; + /// Field id: 514. + int64_t deleted_rows_count = 0; + /// Field id: 524. + int64_t replaced_rows_count = 0; + /// Field id: 516. Minimum data sequence number of live entries. + int64_t min_sequence_number = 0; + + static std::shared_ptr Type(); +}; + +} // namespace iceberg diff --git a/src/iceberg/snapshot.cc b/src/iceberg/snapshot.cc index f791c3724..a77901168 100644 --- a/src/iceberg/snapshot.cc +++ b/src/iceberg/snapshot.cc @@ -27,6 +27,7 @@ #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_list.h" #include "iceberg/manifest/manifest_reader.h" +#include "iceberg/table_metadata.h" #include "iceberg/util/macros.h" #include "iceberg/util/string_util.h" @@ -165,6 +166,35 @@ Result> Snapshot::FirstRowId() const { return first_row_i Result> Snapshot::AddedRows() const { return added_rows; } +Status Snapshot::ValidateForFormatVersion(int8_t format_version, + bool is_new_snapshot) const { + if (!root_manifest.has_value()) { + ICEBERG_CHECK( + !is_new_snapshot || format_version < TableMetadata::kMinFormatVersionRootManifest, + "New v4 snapshots require root-manifest instead of manifest-list"); + ICEBERG_CHECK(!manifest_list.empty(), "Snapshot manifest-list cannot be empty"); + return {}; + } + + ICEBERG_CHECK(format_version >= TableMetadata::kMinFormatVersionRootManifest, + "root-manifest requires format version {}", + TableMetadata::kMinFormatVersionRootManifest); + ICEBERG_CHECK(manifest_list.empty(), "root-manifest cannot coexist with manifest-list"); + ICEBERG_CHECK(!root_manifest->empty(), "Snapshot root-manifest cannot be empty"); + ICEBERG_CHECK(sequence_number > 0, "V4 snapshots require a positive sequence-number"); + ICEBERG_CHECK(first_row_id.has_value() && first_row_id.value() >= 0, + "V4 snapshots require a non-negative first-row-id"); + ICEBERG_CHECK(added_rows.has_value() && added_rows.value() >= 0, + "V4 snapshots require non-negative added-rows"); + const auto operation = Operation(); + ICEBERG_CHECK( + operation.has_value() && + (operation == DataOperation::kAppend || operation == DataOperation::kReplace || + operation == DataOperation::kOverwrite || operation == DataOperation::kDelete), + "V4 snapshots require a valid operation in summary"); + return {}; +} + bool Snapshot::Equals(const Snapshot& other) const { if (this == &other) { return true; @@ -173,7 +203,8 @@ bool Snapshot::Equals(const Snapshot& other) const { parent_snapshot_id == other.parent_snapshot_id && sequence_number == other.sequence_number && timestamp_ms == other.timestamp_ms && schema_id == other.schema_id && first_row_id == other.first_row_id && - added_rows == other.added_rows && key_id == other.key_id; + added_rows == other.added_rows && key_id == other.key_id && + root_manifest == other.root_manifest; } Result> Snapshot::Make( @@ -205,6 +236,9 @@ Result> Snapshot::Make( Result SnapshotReader::InitManifestsCache( const Snapshot* snapshot, std::shared_ptr file_io) { + if (snapshot->root_manifest.has_value()) { + return NotSupported("Reading v4 root manifests is not supported yet"); + } if (file_io == nullptr) { return InvalidArgument("Cannot cache manifests: FileIO is null"); } diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h index 54c6fea2d..2f648220a 100644 --- a/src/iceberg/snapshot.h +++ b/src/iceberg/snapshot.h @@ -417,6 +417,17 @@ struct ICEBERG_EXPORT Snapshot { /// encrypted key metadata of the manifest list. Set only for encrypted tables. std::optional key_id; + /// Location of the root manifest for a snapshot written in v4. Mutually + /// exclusive with manifest_list. Older snapshots retain their manifest list + /// when a table is upgraded. + std::optional root_manifest; + + /// \brief Validate snapshot metadata against a table format version. + /// Existing snapshots may retain a legacy manifest list after a v4 upgrade; + /// newly added v4 snapshots must use a root manifest. + Status ValidateForFormatVersion(int8_t format_version, + bool is_new_snapshot = false) const; + /// \brief Create a new Snapshot instance with validation on the inputs. static Result> Make( int64_t sequence_number, int64_t snapshot_id, diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index b41b2abb8..de0eec450 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -194,10 +194,27 @@ std::string ToString(const MetadataLogEntry& entry) { entry.metadata_file); } +Result> TableMetadata::Make( + const iceberg::Schema& schema, const iceberg::PartitionSpec& spec, + const iceberg::SortOrder& sort_order, const std::string& location, + const std::unordered_map& properties) { + auto persisted_properties = properties; + int format_version = kDefaultTableFormatVersion; + if (auto it = persisted_properties.find(TableProperties::kFormatVersion.key()); + it != persisted_properties.end()) { + ICEBERG_ASSIGN_OR_RAISE(format_version, + StringUtils::ParseNumber(it->second)); + persisted_properties.erase(it); + } + return Make(schema, spec, sort_order, location, persisted_properties, format_version); +} + Result> TableMetadata::Make( const iceberg::Schema& schema, const iceberg::PartitionSpec& spec, const iceberg::SortOrder& sort_order, const std::string& location, const std::unordered_map& properties, int format_version) { + ICEBERG_PRECHECK(format_version >= 1 && format_version <= kSupportedTableFormatVersion, + "Unsupported table format version: {}", format_version); for (const auto& [key, _] : properties) { if (TableProperties::reserved_properties().contains(key)) { return InvalidArgument( @@ -297,6 +314,20 @@ Result> TableMetadata::SnapshotById(int64_t snapshot_i return *iter; } +bool TableMetadata::HasV4RootManifest() const { + return std::ranges::any_of(snapshots, [](const auto& snapshot) { + return snapshot && snapshot->root_manifest.has_value(); + }); +} + +Status TableMetadata::ValidateLocation() const { + if (format_version >= kMinFormatVersionOptionalLocation) { + ICEBERG_PRECHECK(location.empty() || !LocationUtil::ParseScheme(location).empty(), + "V4 table location must be absolute: {}", location); + } + return {}; +} + int64_t TableMetadata::NextSequenceNumber() const { return format_version > 1 ? last_sequence_number + 1 : kInitialSequenceNumber; } @@ -556,6 +587,10 @@ Result TableMetadataUtil::NewTableMetadataFilePath(const TableMetad return std::format("{}/{}", LocationUtil::StripTrailingSlash(metadata_location), filename); } else { + ICEBERG_PRECHECK(!meta.location.empty(), + "Cannot generate a metadata file location without a table location; " + "provide a table location or set '{}'", + TableProperties::kWriteMetadataLocation.key()); return std::format("{}/{}/{}", meta.location, kMetadataFolderName, filename); } } @@ -748,7 +783,8 @@ Status TableMetadataBuilder::Impl::AssignUUID(std::string_view uuid) { Status TableMetadataBuilder::Impl::UpgradeFormatVersion(int8_t new_format_version) { // Check that the new format version is supported - if (new_format_version > TableMetadata::kSupportedTableFormatVersion) { + if (new_format_version < 1 || + new_format_version > TableMetadata::kSupportedTableFormatVersion) { return InvalidArgument( "Cannot upgrade table to unsupported format version: v{} (supported: v{})", new_format_version, TableMetadata::kSupportedTableFormatVersion); @@ -765,6 +801,18 @@ Status TableMetadataBuilder::Impl::UpgradeFormatVersion(int8_t new_format_versio return {}; } + // A create transaction may add schemas before selecting the current schema. + // Build() will validate the schema selected after this upgrade. + if (new_format_version >= TableMetadata::kMinFormatVersionConstraints) { + auto current_schema = schemas_by_id_.find(metadata_.current_schema_id); + if (current_schema != schemas_by_id_.end() && + !current_schema->second->IdentifierFieldIds().empty()) { + return NotSupported( + "Cannot upgrade a table with identifier fields to v4: primary-key " + "constraint migration is not supported yet"); + } + } + // Update the format version metadata_.format_version = new_format_version; @@ -1089,6 +1137,8 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr snapsho "Attempting to add a snapshot before a sort order is added"); ICEBERG_CHECK(!snapshots_by_id_.contains(snapshot->snapshot_id), "Snapshot already exists for id: {}", snapshot->snapshot_id); + ICEBERG_RETURN_UNEXPECTED( + snapshot->ValidateForFormatVersion(metadata_.format_version, true)); if (metadata_.format_version != 1 && snapshot->sequence_number <= metadata_.last_sequence_number && snapshot->parent_snapshot_id.has_value()) { @@ -1096,7 +1146,15 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr snapsho "Cannot add snapshot with sequence number {} older than last sequence number {}", snapshot->sequence_number, metadata_.last_sequence_number); } - + if (metadata_.format_version >= TableMetadata::kMinFormatVersionMonotonicTimestamps && + snapshot->parent_snapshot_id.has_value()) { + auto parent = snapshots_by_id_.find(snapshot->parent_snapshot_id.value()); + if (parent != snapshots_by_id_.end() && + snapshot->timestamp_ms <= parent->second->timestamp_ms) { + return ValidationFailed( + "V4 snapshot timestamp must be greater than its parent timestamp"); + } + } metadata_.last_sequence_number = snapshot->sequence_number; metadata_.snapshots.push_back(snapshot); snapshots_by_id_.emplace(snapshot->snapshot_id, snapshot); @@ -1373,6 +1431,11 @@ Result> TableMetadataBuilder::Impl::UpdateSnapshot Result> TableMetadataBuilder::Impl::Build() { // 1. Validate metadata consistency through TableMetadata#Validate + ICEBERG_PRECHECK( + metadata_.format_version >= 1 && + metadata_.format_version <= TableMetadata::kSupportedTableFormatVersion, + "Unsupported table format version: {}", metadata_.format_version); + ICEBERG_RETURN_UNEXPECTED(metadata_.ValidateLocation()); // 2. Update last_updated_ms if there are changes if (metadata_.last_updated_ms == kInvalidLastUpdatedMs) { @@ -1387,6 +1450,14 @@ Result> TableMetadataBuilder::Impl::Build() { "Current schema ID {} not found in schemas", current_schema_id); { const auto& current_schema = schema_it->second; + if (metadata_.format_version >= TableMetadata::kMinFormatVersionConstraints && + std::ranges::any_of(metadata_.schemas, [](const auto& schema) { + return !schema->IdentifierFieldIds().empty(); + })) { + return NotSupported( + "V4 primary-key constraints are not supported yet; " + "identifier-field-ids cannot be written in v4 schemas"); + } auto spec_it = specs_by_id_.find(metadata_.default_spec_id); ICEBERG_PRECHECK(spec_it != specs_by_id_.end(), diff --git a/src/iceberg/table_metadata.h b/src/iceberg/table_metadata.h index 6e2d9fe26..aa77af30d 100644 --- a/src/iceberg/table_metadata.h +++ b/src/iceberg/table_metadata.h @@ -72,7 +72,13 @@ struct ICEBERG_EXPORT MetadataLogEntry { /// (4) Map struct ICEBERG_EXPORT TableMetadata { static constexpr int8_t kDefaultTableFormatVersion = 2; - static constexpr int8_t kSupportedTableFormatVersion = 3; + /// Highest supported table metadata version. V4 manifest tree I/O is not yet + /// implemented. + static constexpr int8_t kSupportedTableFormatVersion = 4; + static constexpr int8_t kMinFormatVersionOptionalLocation = 4; + static constexpr int8_t kMinFormatVersionRootManifest = 4; + static constexpr int8_t kMinFormatVersionMonotonicTimestamps = 4; + static constexpr int8_t kMinFormatVersionConstraints = 4; static constexpr int8_t kMinFormatVersionRowLineage = 3; static constexpr int8_t kMinFormatVersionDefaultValues = 3; static constexpr int64_t kInitialSequenceNumber = 0; @@ -87,7 +93,8 @@ struct ICEBERG_EXPORT TableMetadata { int8_t format_version; /// A UUID that identifies the table std::string table_uuid; - /// The table's base location + /// The table's base location. Empty represents an omitted location in v4 + /// metadata; a catalog must provide the location before resolving relative paths. std::string location; /// The table's highest assigned sequence number int64_t last_sequence_number; @@ -135,8 +142,12 @@ struct ICEBERG_EXPORT TableMetadata { static Result> Make( const iceberg::Schema& schema, const iceberg::PartitionSpec& spec, const iceberg::SortOrder& sort_order, const std::string& location, - const std::unordered_map& properties, - int format_version = kDefaultTableFormatVersion); + const std::unordered_map& properties); + + static Result> Make( + const iceberg::Schema& schema, const iceberg::PartitionSpec& spec, + const iceberg::SortOrder& sort_order, const std::string& location, + const std::unordered_map& properties, int format_version); /// \brief Get the current schema, return NotFoundError if not found /// \note The returned schema is guaranteed to be not null @@ -166,6 +177,13 @@ struct ICEBERG_EXPORT TableMetadata { /// \brief Get the next sequence number int64_t NextSequenceNumber() const; + /// \brief Whether any snapshot references a v4 root manifest. + bool HasV4RootManifest() const; + + /// \brief Validate the table location for this format version. + /// V4 locations may be absent, but a present location must be absolute. + Status ValidateLocation() const; + ICEBERG_EXPORT friend bool operator==(const TableMetadata& lhs, const TableMetadata& rhs); }; diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index b2fd4f9be..1270ca273 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -84,6 +84,9 @@ add_iceberg_test(schema_test transform_test.cc type_test.cc) +add_iceberg_test(v4_types_test SOURCES v4_types_test.cc) +add_iceberg_test(v4_table_metadata_test SOURCES v4_table_metadata_test.cc) + add_iceberg_test(table_test SOURCES catalog_util_test.cc diff --git a/src/iceberg/test/expire_snapshots_test.cc b/src/iceberg/test/expire_snapshots_test.cc index 9695b4a85..09054b54d 100644 --- a/src/iceberg/test/expire_snapshots_test.cc +++ b/src/iceberg/test/expire_snapshots_test.cc @@ -30,6 +30,7 @@ #include #include "iceberg/avro/avro_register.h" +#include "iceberg/logging/logger.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_writer.h" #include "iceberg/partition_spec.h" @@ -39,6 +40,7 @@ #include "iceberg/table_metadata.h" #include "iceberg/table_properties.h" #include "iceberg/test/executor.h" +#include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/test/mock_catalog.h" #include "iceberg/test/update_test_base.h" @@ -197,6 +199,28 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase { RewriteTable(std::move(metadata)); } + void UseV4Root(TableMetadata& metadata) const { + metadata.format_version = 4; + // These fixtures model v4 schemas without primary keys; production identifier + // migration remains unsupported and serialization must not drop identifiers. + for (auto& schema : metadata.schemas) { + if (!schema->IdentifierFieldIds().empty()) { + schema = std::make_shared( + std::vector(schema->fields().begin(), schema->fields().end()), + schema->schema_id()); + } + } + metadata.location = "file:" + table_location_; + // Keep the mock filesystem's metadata writes at its existing absolute path. + metadata.properties.Set(TableProperties::kWriteMetadataLocation, + table_location_ + "/metadata"); + auto& current = *metadata.snapshots.back(); + current.manifest_list.clear(); + current.root_manifest = table_location_ + "/metadata/current-root.avro"; + current.first_row_id = 0; + current.added_rows = 0; + } + void RewriteTable(std::shared_ptr metadata) { ASSERT_NE(metadata, nullptr); @@ -224,6 +248,169 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase { } }; +class ExpireSnapshotsV4CleanupTest : public ExpireSnapshotsCleanupTest, + public ::testing::WithParamInterface { +}; + +TEST_P(ExpireSnapshotsV4CleanupTest, RejectsCleanupBeforeCommit) { + const auto data_path = table_location_ + "/data/shared.parquet"; + const auto manifest_path = table_location_ + "/metadata/shared-manifest.avro"; + const auto expired_list = table_location_ + "/metadata/expired-list.avro"; + ASSERT_THAT(file_io_->WriteFile(data_path, "data"), IsOk()); + auto manifest = + WriteDataManifest(manifest_path, kExpiredSnapshotId, + {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, + kExpiredSequenceNumber, MakeDataFile(data_path))}); + WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, + {manifest}); + + auto metadata = ReloadMetadata(); + metadata->snapshots.front()->manifest_list = expired_list; + UseV4Root(*metadata); + RewriteTable(std::move(metadata)); + const auto metadata_location = std::string(table_->metadata_file_location()); + + // The opaque v4 root may still reference the legacy manifest. Exercise both + // incremental (by age) and reachable (by explicit ID) cleanup selection. + for (bool explicit_id : {false, true}) { + SCOPED_TRACE(explicit_id); + std::vector deleted_files; + ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots()); + if (explicit_id) { + update->ExpireSnapshotId(kExpiredSnapshotId); + } else { + update->ExpireOlderThan( + UnixMsFromTimePointMs(table_->metadata()->snapshots.back()->timestamp_ms)); + } + update->CleanupLevel(GetParam()).DeleteWith([&](const std::string& path) { + deleted_files.push_back(path); + }); + EXPECT_THAT(update->Apply(), IsError(ErrorKind::kNotSupported)); + EXPECT_THAT(update->Commit(), IsError(ErrorKind::kNotSupported)); + EXPECT_TRUE(deleted_files.empty()); + EXPECT_TRUE(ReloadMetadata()->SnapshotById(kExpiredSnapshotId).has_value()); + EXPECT_EQ(table_->metadata_file_location(), metadata_location); + for (const auto& path : {data_path, manifest_path, expired_list}) { + EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk()); + } + } +} + +INSTANTIATE_TEST_SUITE_P(FileCleanupLevels, ExpireSnapshotsV4CleanupTest, + ::testing::Values(CleanupLevel::kAll, + CleanupLevel::kMetadataOnly)); + +TEST_F(ExpireSnapshotsCleanupTest, V4SupportsNoOpAndExpirationWithoutFileCleanup) { + const auto expired_list = table_location_ + "/metadata/expired-list.avro"; + WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, {}); + auto metadata = ReloadMetadata(); + metadata->snapshots.front()->manifest_list = expired_list; + UseV4Root(*metadata); + RewriteTable(std::move(metadata)); + + ICEBERG_UNWRAP_OR_FAIL(auto noop, table_->NewExpireSnapshots()); + noop->RetainLast(2); + EXPECT_THAT(noop->Commit(), IsOk()); + + std::vector deleted_files; + ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots()); + update->ExpireSnapshotId(kExpiredSnapshotId) + .CleanupLevel(CleanupLevel::kNone) + .DeleteWith([&](const std::string& path) { deleted_files.push_back(path); }); + ASSERT_THAT(update->Commit(), IsOk()); + EXPECT_TRUE(deleted_files.empty()); + metadata = ReloadMetadata(); + EXPECT_FALSE(metadata->SnapshotById(kExpiredSnapshotId).has_value()); + EXPECT_TRUE(metadata->SnapshotById(kCurrentSnapshotId).has_value()); + EXPECT_THAT(file_io_->ReadFile(expired_list, std::nullopt), IsOk()); +} + +TEST_F(ExpireSnapshotsCleanupTest, V4RejectsSchemaSpecPruningBeforeCommit) { + auto metadata = ReloadMetadata(); + UseV4Root(*metadata); + RewriteTable(std::move(metadata)); + const auto location = std::string(table_->metadata_file_location()); + for (bool expire_snapshot : {false, true}) { + SCOPED_TRACE(expire_snapshot); + std::vector deleted_files; + ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots()); + update->CleanupLevel(CleanupLevel::kNone) + .CleanExpiredMetadata(true) + .DeleteWith([&](const std::string& path) { deleted_files.push_back(path); }); + if (expire_snapshot) { + update->ExpireSnapshotId(kExpiredSnapshotId); + } else { + update->RetainLast(2); + } + auto result = update->Apply(); + EXPECT_THAT(result, IsError(ErrorKind::kNotSupported)); + EXPECT_THAT(result, HasErrorMessage("disable CleanExpiredMetadata")); + EXPECT_THAT(update->Commit(), IsError(ErrorKind::kNotSupported)); + EXPECT_TRUE(deleted_files.empty()); + EXPECT_EQ(table_->metadata_file_location(), location); + EXPECT_TRUE(ReloadMetadata()->SnapshotById(kExpiredSnapshotId).has_value()); + } +} + +TEST_P(ExpireSnapshotsV4CleanupTest, SkipsCleanupWhenCommittedMetadataAddsV4Root) { + const auto data_path = table_location_ + "/data/expired.parquet"; + const auto manifest_path = table_location_ + "/metadata/expired-manifest.avro"; + const auto expired_list = table_location_ + "/metadata/expired-list.avro"; + const auto current_list = table_location_ + "/metadata/current-list.avro"; + ASSERT_THAT(file_io_->WriteFile(data_path, "data"), IsOk()); + auto manifest = + WriteDataManifest(manifest_path, kExpiredSnapshotId, + {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, + kExpiredSequenceNumber, MakeDataFile(data_path))}); + WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, + {manifest}); + WriteManifestList(current_list, kCurrentSnapshotId, kExpiredSnapshotId, + kCurrentSequenceNumber, {}); + RewriteTableWithManifestLists(expired_list, current_list); + + auto mock = std::make_shared<::testing::NiceMock>(); + EXPECT_CALL(*mock, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillOnce([this](const auto& name, const auto& requirements, + const auto& updates) -> Result> { + ICEBERG_ASSIGN_OR_RAISE(auto committed, + catalog_->UpdateTable(name, requirements, updates)); + UseV4Root(*committed->metadata()); + ICEBERG_RETURN_UNEXPECTED(TableMetadataUtil::Write( + *file_io_, std::string(committed->metadata_file_location()), + *committed->metadata())); + return committed; + }); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), file_io_, mock)); + std::vector deleted_files; + ICEBERG_UNWRAP_OR_FAIL(auto update, table->NewExpireSnapshots()); + update->ExpireSnapshotId(kExpiredSnapshotId) + .CleanupLevel(GetParam()) + .DeleteWith([&](const std::string& path) { deleted_files.push_back(path); }); + auto logger = std::make_shared(); + ScopedLogger scoped_logger(logger); + ASSERT_THAT(update->Commit(), IsOk()); + EXPECT_TRUE(deleted_files.empty()); + auto committed = ReloadMetadata(); + EXPECT_FALSE(committed->SnapshotById(kExpiredSnapshotId).has_value()); + EXPECT_TRUE(committed->snapshots.back()->root_manifest.has_value()); + for (const auto& path : {expired_list, manifest_path, data_path}) { + EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk()); + } + EXPECT_THAT(logger->records(), + ::testing::Contains(::testing::AllOf( + ::testing::Field(&LogMessage::level, LogLevel::kWarn), + ::testing::Field( + &LogMessage::message, + ::testing::HasSubstr("expired snapshot files were retained"))))); + for (const auto& record : logger->records()) { + EXPECT_THAT(record.message, + ::testing::Not(::testing::HasSubstr("Update finalization failed"))); + } +} + TEST_F(ExpireSnapshotsTest, DefaultExpireByAge) { ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots()); ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply()); diff --git a/src/iceberg/test/json_serde_test.cc b/src/iceberg/test/json_serde_test.cc index 13b28a7bb..5770ca190 100644 --- a/src/iceberg/test/json_serde_test.cc +++ b/src/iceberg/test/json_serde_test.cc @@ -44,6 +44,7 @@ #include "iceberg/sort_field.h" #include "iceberg/sort_order.h" #include "iceberg/statistics_file.h" +#include "iceberg/table_metadata.h" #include "iceberg/table_requirement.h" #include "iceberg/table_scan.h" #include "iceberg/table_update.h" @@ -687,16 +688,50 @@ TEST(JsonInternalTest, TableUpdateAssignUUID) { } TEST(JsonInternalTest, TableUpdateUpgradeFormatVersion) { - table::UpgradeFormatVersion update(2); - nlohmann::json expected = - R"({"action":"upgrade-format-version","format-version":2})"_json; + for (int8_t version : + std::array{1, 2, 3, 4, 0, 5, std::numeric_limits::min(), + std::numeric_limits::max()}) { + SCOPED_TRACE(version); + table::UpgradeFormatVersion update(version); + nlohmann::json expected = {{"action", "upgrade-format-version"}, + {"format-version", version}}; + + ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(update)); + EXPECT_EQ(json, expected); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableUpdateFromJson(expected)); + const auto& actual = + internal::checked_cast(*parsed); + EXPECT_EQ(actual.format_version(), version); + EXPECT_EQ(actual, update); + + if (version < 1 || version > TableMetadata::kSupportedTableFormatVersion) { + auto builder = TableMetadataBuilder::BuildFromEmpty(2); + parsed->ApplyTo(*builder); + auto result = builder->Build(); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, HasErrorMessage( + "Cannot upgrade table to unsupported format version: v" + + std::to_string(version))); + EXPECT_TRUE(builder->changes().empty()); + } + } +} - ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(update)); - EXPECT_EQ(json, expected); - auto parsed = TableUpdateFromJson(expected); - ASSERT_THAT(parsed, IsOk()); - EXPECT_EQ(*internal::checked_cast(parsed.value().get()), - update); +TEST(JsonInternalTest, TableUpdateUpgradeFormatVersionRejectsInvalidIntegers) { + for (const auto& value : + {nlohmann::json(128), nlohmann::json(-129), nlohmann::json(260), + nlohmann::json(-252), nlohmann::json(4294967300LL), nlohmann::json(-4294967292LL), + nlohmann::json(2147483648LL), nlohmann::json(-2147483649LL), + nlohmann::json(std::numeric_limits::max()), nlohmann::json(4.0), + nlohmann::json(4.5), nlohmann::json("4"), nlohmann::json(true), + nlohmann::json(nullptr)}) { + SCOPED_TRACE(value.dump()); + nlohmann::json json = {{"action", "upgrade-format-version"}, + {"format-version", value}}; + EXPECT_THAT(TableUpdateFromJson(json), IsError(ErrorKind::kJsonParseError)); + } + EXPECT_THAT(TableUpdateFromJson({{"action", "upgrade-format-version"}}), + IsError(ErrorKind::kJsonParseError)); } TEST(JsonInternalTest, TableUpdateAddSchema) { diff --git a/src/iceberg/test/location_provider_test.cc b/src/iceberg/test/location_provider_test.cc index 268199b3b..8b1f3062a 100644 --- a/src/iceberg/test/location_provider_test.cc +++ b/src/iceberg/test/location_provider_test.cc @@ -74,6 +74,35 @@ TEST_F(LocationProviderTest, DefaultLocationProviderWithCustomDataLocation) { EXPECT_EQ("new_location/my_file", location); } +TEST_F(LocationProviderTest, MissingTableLocationRequiresAnExplicitDataLocation) { + auto missing = LocationProvider::Make("", properties_); + EXPECT_THAT(missing, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(missing, HasErrorMessage("absolute URI in 'write.data.path'")); + for (const auto* data_location : {"data", "/data"}) { + properties_.Set(TableProperties::kWriteDataLocation, std::string(data_location)); + EXPECT_THAT(LocationProvider::Make("", properties_), + IsError(ErrorKind::kInvalidArgument)); + } + properties_.Set(TableProperties::kWriteDataLocation, std::string("s3://data/table")); + ICEBERG_UNWRAP_OR_FAIL(auto provider, LocationProvider::Make("", properties_)); + EXPECT_EQ(provider->NewDataLocation("f.parquet"), "s3://data/table/f.parquet"); +} + +TEST_F(LocationProviderTest, ObjectStoreRequiresTableLocationForPathContext) { + properties_.Set(TableProperties::kObjectStoreEnabled, true); + auto missing = LocationProvider::Make("", properties_); + EXPECT_THAT(missing, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(missing, HasErrorMessage( + "Object-store location provider requires a table location")); + properties_.Set(TableProperties::kWriteDataLocation, std::string("s3://shared-data")); + EXPECT_THAT(LocationProvider::Make("", properties_), + IsError(ErrorKind::kInvalidArgument)); + ICEBERG_UNWRAP_OR_FAIL(auto provider, + LocationProvider::Make("s3://bucket/db/table", properties_)); + EXPECT_THAT(provider->NewDataLocation("f.parquet"), + testing::EndsWith("/db/table/f.parquet")); +} + TEST_F(LocationProviderTest, ObjectStorageLocationProvider) { std::ignore = properties_.Set(TableProperties::kObjectStoreEnabled, true); ICEBERG_UNWRAP_OR_FAIL(auto provider, diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc index ae5ba1a74..3470a2193 100644 --- a/src/iceberg/test/rest_json_serde_test.cc +++ b/src/iceberg/test/rest_json_serde_test.cc @@ -1265,7 +1265,7 @@ INSTANTIATE_TEST_SUITE_P( .test_name = "InvalidMetadataContent", .invalid_json_str = R"({"metadata":{"format-version":"invalid"}})", .expected_error_message = - "Failed to parse 'format-version' (JSON value of type string)"}, + "'format-version' must be an integer, but is string"}, LoadTableResultInvalidParam{ .test_name = "LabelsFieldsNotArray", .invalid_json_str = LoadTableJsonWithLabels(R"({"fields":"oops"})"), diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc index dea10a01c..8c0f03b8d 100644 --- a/src/iceberg/test/snapshots_table_test.cc +++ b/src/iceberg/test/snapshots_table_test.cc @@ -86,7 +86,7 @@ TEST_F(SnapshotsTableTest, Scan) { // Row and column counts. EXPECT_EQ(batch->num_rows(), 2); - EXPECT_EQ(batch->num_columns(), 6); + EXPECT_EQ(batch->num_columns(), 7); // Column 0: committed_at (timestamptz) — microseconds since epoch. auto committed_at = std::static_pointer_cast<::arrow::TimestampArray>(batch->column(0)); @@ -132,6 +132,56 @@ TEST_F(SnapshotsTableTest, Scan) { second_summary, ::testing::Not(::testing::Contains(::testing::Pair("operation", "append")))); EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2"))); + + // Legacy snapshots have no root manifest. + EXPECT_EQ(batch->column(6)->null_count(), 2); +} + +TEST_F(SnapshotsTableTest, ExposesRootManifestAndNullLegacyManifestList) { + auto& metadata = *table_->metadata(); + metadata.format_version = 4; + auto& snapshot = *metadata.snapshots.back(); + snapshot.manifest_list.clear(); + snapshot.root_manifest = "s3://bucket/metadata/root.avro"; + snapshot.first_row_id = 0; + snapshot.added_rows = 0; + + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table_)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + const auto& batch = batches.front(); + auto lists = std::static_pointer_cast<::arrow::StringArray>( + batch->GetColumnByName("manifest_list")); + auto roots = std::static_pointer_cast<::arrow::StringArray>( + batch->GetColumnByName("root_manifest")); + ASSERT_NE(lists, nullptr); + ASSERT_NE(roots, nullptr); + EXPECT_EQ(lists->GetString(0), "file:/tmp/manifest1.avro"); + EXPECT_TRUE(lists->IsNull(1)); + EXPECT_TRUE(roots->IsNull(0)); + EXPECT_EQ(roots->GetString(1), snapshot.root_manifest.value()); + + auto projection = std::make_shared( + std::vector{SchemaField::MakeOptional(9, "root_manifest", string()), + SchemaField::MakeOptional(5, "manifest_list", string())}); + ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan()); + ICEBERG_UNWRAP_OR_FAIL(auto projected_stream, + BuildStream(builder->Project(projection))); + ICEBERG_UNWRAP_OR_FAIL(auto projected_batches, + ReadAllBatches(std::move(projected_stream))); + ASSERT_EQ(projected_batches.size(), 1); + const auto& projected = projected_batches.front(); + ASSERT_EQ(projected->num_columns(), 2); + EXPECT_EQ(projected->column_name(0), "root_manifest"); + EXPECT_EQ(projected->column_name(1), "manifest_list"); + auto projected_roots = + std::static_pointer_cast<::arrow::StringArray>(projected->column(0)); + auto projected_lists = + std::static_pointer_cast<::arrow::StringArray>(projected->column(1)); + EXPECT_TRUE(projected_roots->IsNull(0)); + EXPECT_EQ(projected_roots->GetString(1), snapshot.root_manifest.value()); + EXPECT_EQ(projected_lists->GetString(0), "file:/tmp/manifest1.avro"); + EXPECT_TRUE(projected_lists->IsNull(1)); } TEST_F(SnapshotsTableTest, ScanSelectedColumns) { @@ -188,7 +238,7 @@ TEST_F(SnapshotsTableTest, ScanAllColumnsSelection) { ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); ASSERT_EQ(batches.size(), 1); EXPECT_EQ(batches.front()->num_rows(), 2); - EXPECT_EQ(batches.front()->num_columns(), 6); + EXPECT_EQ(batches.front()->num_columns(), 7); } TEST_F(SnapshotsTableTest, ScanBuildsRepeatedly) { @@ -405,7 +455,8 @@ TEST_F(SnapshotsTableTest, CanonicalSchema) { SchemaField::MakeOptional( 6, "summary", std::make_shared(SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); + SchemaField::MakeRequired(8, "value", string()))), + SchemaField::MakeOptional(9, "root_manifest", string())}); EXPECT_EQ(*SnapshotsTable::CanonicalSchema(), *expected); EXPECT_EQ(*snapshots_table_->schema(), *expected); } diff --git a/src/iceberg/test/table_metadata_builder_test.cc b/src/iceberg/test/table_metadata_builder_test.cc index 5a9681add..959f2dbfd 100644 --- a/src/iceberg/test/table_metadata_builder_test.cc +++ b/src/iceberg/test/table_metadata_builder_test.cc @@ -17,6 +17,8 @@ * under the License. */ +#include +#include #include #include #include @@ -432,7 +434,7 @@ TEST(TableMetadataBuilderTest, UpgradeFormatVersion) { // Unsupported format version should fail builder = TableMetadataBuilder::BuildFromEmpty(3); - builder->UpgradeFormatVersion(4); + builder->UpgradeFormatVersion(5); EXPECT_THAT(builder->Build(), HasErrorMessage("Cannot upgrade table to unsupported format version")); @@ -442,6 +444,44 @@ TEST(TableMetadataBuilderTest, UpgradeFormatVersion) { EXPECT_THAT(builder->Build(), HasErrorMessage("Cannot downgrade")); } +TEST(TableMetadataBuilderTest, UpgradeFormatVersionRejectsUnsupportedVersions) { + auto base = CreateBaseMetadata(); + for (int8_t version : std::vector{0, 5, -1, std::numeric_limits::min(), + std::numeric_limits::max()}) { + SCOPED_TRACE(version); + auto empty_builder = TableMetadataBuilder::BuildFromEmpty(version); + auto empty_result = empty_builder->Build(); + EXPECT_THAT(empty_result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(empty_result, HasErrorMessage("Unsupported table format version: " + + std::to_string(version))); + + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->UpgradeFormatVersion(version); + auto result = builder->Build(); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, + HasErrorMessage("Cannot upgrade table to unsupported format version: v" + + std::to_string(version))); + EXPECT_TRUE(builder->changes().empty()); + + table::UpgradeFormatVersion update(version); + auto cloned = update.Clone(); + EXPECT_EQ(*cloned, update); + const std::array updates{&update, cloned.get()}; + for (const auto* candidate : updates) { + auto update_builder = TableMetadataBuilder::BuildFrom(base.get()); + candidate->ApplyTo(*update_builder); + auto update_result = update_builder->Build(); + EXPECT_THAT(update_result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT( + update_result, + HasErrorMessage("Cannot upgrade table to unsupported format version: v" + + std::to_string(version))); + EXPECT_TRUE(update_builder->changes().empty()); + } + } +} + // Test AddSortOrder TEST(TableMetadataBuilderTest, AddSortOrderBasic) { auto base = CreateBaseMetadata(); @@ -1292,8 +1332,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshotRef) { auto builder = TableMetadataBuilder::BuildFrom(base.get()); // Add multiple snapshots - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 1})); - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 2})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"})); // Add multiple refs ICEBERG_UNWRAP_OR_FAIL(auto ref1, SnapshotRef::MakeBranch(1)); @@ -1316,7 +1358,8 @@ TEST(TableMetadataBuilderTest, SetRefRejectsTagForMainBranch) { auto base = CreateBaseMetadata(); auto builder = TableMetadataBuilder::BuildFrom(base.get()); - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 1})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"})); ICEBERG_UNWRAP_OR_FAIL(auto main_tag, SnapshotRef::MakeTag(1)); builder->SetRef(std::string(SnapshotRef::kMainBranch), std::move(main_tag)); @@ -1332,7 +1375,10 @@ TEST(TableMetadataBuilderTest, SetMainRefToAddedSnapshotUsesSnapshotTimestampFor auto snapshot_time = TimePointMsFromUnixMs(123456789); builder->AddSnapshot(std::make_shared( - Snapshot{.snapshot_id = 1, .sequence_number = 1, .timestamp_ms = snapshot_time})); + Snapshot{.snapshot_id = 1, + .sequence_number = 1, + .timestamp_ms = snapshot_time, + .manifest_list = "s3://bucket/manifest-list-1.avro"})); ICEBERG_UNWRAP_OR_FAIL(auto main_branch, SnapshotRef::MakeBranch(1)); builder->SetRef(std::string(SnapshotRef::kMainBranch), std::move(main_branch)); @@ -1350,8 +1396,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshot) { auto builder = TableMetadataBuilder::BuildFrom(base.get()); // Add multiple snapshots - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 1})); - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 2})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"})); ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build()); ASSERT_EQ(metadata->snapshots.size(), 2); @@ -1370,8 +1418,10 @@ TEST(TableMetadataBuilderTest, RemoveSnapshotNotExist) { auto builder = TableMetadataBuilder::BuildFrom(base.get()); // Add multiple snapshots - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 1})); - builder->AddSnapshot(std::make_shared(Snapshot{.snapshot_id = 2})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 1, .manifest_list = "s3://bucket/manifest-list-1.avro"})); + builder->AddSnapshot(std::make_shared( + Snapshot{.snapshot_id = 2, .manifest_list = "s3://bucket/manifest-list-2.avro"})); ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build()); ASSERT_EQ(metadata->snapshots.size(), 2); diff --git a/src/iceberg/test/update_schema_test.cc b/src/iceberg/test/update_schema_test.cc index 4057c52c8..474699135 100644 --- a/src/iceberg/test/update_schema_test.cc +++ b/src/iceberg/test/update_schema_test.cc @@ -159,7 +159,7 @@ class UpdateSchemaDefaultValueTest : public UpdateSchemaTest { ICEBERG_UNWRAP_OR_FAIL(auto metadata, ReadTableMetadataFromResource("TableMetadataV2Valid.json")); - metadata->format_version = TableMetadata::kSupportedTableFormatVersion; + metadata->format_version = TableMetadata::kMinFormatVersionDefaultValues; metadata->next_row_id = TableMetadata::kInitialRowId; RegisterTable(std::move(metadata)); } diff --git a/src/iceberg/test/v4_table_metadata_test.cc b/src/iceberg/test/v4_table_metadata_test.cc new file mode 100644 index 000000000..c246f4882 --- /dev/null +++ b/src/iceberg/test/v4_table_metadata_test.cc @@ -0,0 +1,642 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "iceberg/catalog/memory/in_memory_catalog.h" +#include "iceberg/file_io.h" +#include "iceberg/json_serde_internal.h" +#include "iceberg/location_provider.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/snapshot.h" +#include "iceberg/sort_order.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/table_update.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/transaction.h" +#include "iceberg/type.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/update_properties.h" + +namespace iceberg { +namespace { + +std::shared_ptr TestSchema() { + return std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64())}, 0); +} + +Result> NewMetadata(int version = 4) { + return TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(), + *SortOrder::Unsorted(), "s3://bucket/table", {}, version); +} + +// A complete v4 metadata fixture independent of the builder and serializer. +nlohmann::json MetadataJson() { + return nlohmann::json::parse(R"({ + "format-version": 4, + "table-uuid": "c9dc1a9f-6564-42ec-9a87-1db146e18019", + "location": "s3://bucket/table", + "last-sequence-number": 0, + "last-updated-ms": 1000, + "last-column-id": 1, + "schemas": [{"type": "struct", "schema-id": 0, "fields": [ + {"id": 1, "name": "id", "type": "long", "required": true} + ]}], + "current-schema-id": 0, + "partition-specs": [{"spec-id": 0, "fields": []}], + "default-spec-id": 0, + "last-partition-id": 999, + "sort-orders": [{"order-id": 0, "fields": []}], + "default-sort-order-id": 0, + "next-row-id": 0, + "snapshots": [] + })"); +} + +nlohmann::json RootSnapshotJson() { + return nlohmann::json::parse(R"({ + "snapshot-id": 1, + "sequence-number": 1, + "timestamp-ms": 1000, + "root-manifest": "metadata/root.avro", + "summary": {"operation": "append"}, + "schema-id": 0, + "first-row-id": 0, + "added-rows": 10 + })"); +} + +TEST(V4TableMetadataTest, CreateAndRoundTrip) { + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata()); + EXPECT_EQ(metadata->format_version, 4); + EXPECT_EQ(metadata->next_row_id, 0); + EXPECT_TRUE(metadata->snapshots.empty()); + EXPECT_FALSE(metadata->HasV4RootManifest()); + EXPECT_FALSE(metadata->table_uuid.empty()); + ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(*metadata)); + EXPECT_EQ(json["format-version"], 4); + EXPECT_EQ(json["next-row-id"], 0); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json)); + EXPECT_EQ(*metadata, *parsed); +} + +TEST(V4TableMetadataTest, SerializationRejectsNullSnapshots) { + auto snapshot_json = RootSnapshotJson(); + snapshot_json.erase("root-manifest"); + snapshot_json["manifest-list"] = "s3://bucket/metadata/legacy.avro"; + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(snapshot_json)); + auto valid = std::shared_ptr(std::move(snapshot)); + for (int version : {1, 2, 3, 4}) { + SCOPED_TRACE(version); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(version)); + for (bool null_first : {false, true}) { + metadata->snapshots = null_first + ? std::vector>{nullptr, valid} + : std::vector>{valid, nullptr}; + EXPECT_FALSE(metadata->HasV4RootManifest()); + auto result = ToJson(*metadata); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("Cannot write null snapshot")); + } + } +} + +TEST(V4TableMetadataTest, CreateFromFormatVersionProperty) { + ICEBERG_UNWRAP_OR_FAIL( + auto metadata, TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(), + *SortOrder::Unsorted(), "s3://bucket/table", + {{"format-version", "4"}, {"owner", "test"}})); + EXPECT_EQ(metadata->format_version, 4); + EXPECT_FALSE(metadata->properties.configs().contains("format-version")); + EXPECT_EQ(metadata->properties.configs().at("owner"), "test"); + ICEBERG_UNWRAP_OR_FAIL(auto defaults, + NewMetadata(TableMetadata::kDefaultTableFormatVersion)); + EXPECT_EQ(defaults->format_version, 2); +} + +TEST(V4TableMetadataTest, RejectUnsupportedVersionsWithoutNarrowing) { + for (int version : {0, -1, 5, 260}) { + SCOPED_TRACE(version); + EXPECT_THAT(NewMetadata(version), IsError(ErrorKind::kInvalidArgument)); + auto json = MetadataJson(); + json["format-version"] = version; + EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("unsupported version")); + } + auto builder = TableMetadataBuilder::BuildFromEmpty(5); + EXPECT_THAT(builder->Build(), HasErrorMessage("Unsupported table format version")); + for (const auto& version : std::vector{4294967300LL, 4.5, true}) { + auto json = MetadataJson(); + json["format-version"] = version; + EXPECT_FALSE(TableMetadataFromJson(json).has_value()); + } +} + +TEST(V4TableMetadataTest, UpgradePreservesLegacySnapshots) { + for (int version : {1, 2, 3}) { + SCOPED_TRACE(version); + ICEBERG_UNWRAP_OR_FAIL(auto base, NewMetadata(version)); + auto legacy_json = RootSnapshotJson(); + legacy_json.erase("root-manifest"); + legacy_json["manifest-list"] = "s3://bucket/table/metadata/legacy.avro"; + ICEBERG_UNWRAP_OR_FAIL(auto legacy_snapshot, SnapshotFromJson(legacy_json)); + base->snapshots.push_back(std::move(legacy_snapshot)); + base->next_row_id = 10; + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->UpgradeFormatVersion(4); + ICEBERG_UNWRAP_OR_FAIL(auto upgraded, builder->Build()); + EXPECT_EQ(upgraded->format_version, 4); + EXPECT_EQ(upgraded->next_row_id, 10); + EXPECT_EQ(upgraded->table_uuid, base->table_uuid); + ASSERT_EQ(upgraded->snapshots.size(), 1); + EXPECT_EQ(upgraded->snapshots.front()->manifest_list, + legacy_json["manifest-list"].get()); + EXPECT_FALSE(upgraded->snapshots.front()->root_manifest.has_value()); + ASSERT_EQ(builder->changes().size(), 1); + EXPECT_EQ(builder->changes().front()->kind(), + TableUpdate::Kind::kUpgradeFormatVersion); + ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(*upgraded)); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json)); + EXPECT_EQ(*parsed, *upgraded); + auto downgrade = TableMetadataBuilder::BuildFrom(upgraded.get()); + downgrade->UpgradeFormatVersion(3); + EXPECT_THAT(downgrade->Build(), HasErrorMessage("Cannot downgrade")); + } +} + +TEST(V4TableMetadataTest, OptionalLocation) { + auto json = MetadataJson(); + json.erase("location"); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(json)); + EXPECT_TRUE(parsed->location.empty()); + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*parsed)); + EXPECT_FALSE(serialized.contains("location")); + auto builder = TableMetadataBuilder::BuildFrom(parsed.get()); + builder->SetLocation("s3://catalog/table"); + ICEBERG_UNWRAP_OR_FAIL(auto located, builder->Build()); + EXPECT_EQ(located->location, "s3://catalog/table"); + for (int version : {1, 2, 3}) { + json["format-version"] = version; + EXPECT_FALSE(TableMetadataFromJson(json).has_value()); + } +} + +TEST(V4TableMetadataTest, RejectRelativeTableLocationAndMissingRequiredFields) { + auto json = MetadataJson(); + json["location"] = "relative/table"; + EXPECT_THAT(TableMetadataFromJson(json), IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("location must be absolute")); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata()); + auto builder = TableMetadataBuilder::BuildFrom(metadata.get()); + builder->SetLocation("relative/table"); + EXPECT_THAT(builder->Build(), HasErrorMessage("location must be absolute")); + metadata->location = "relative/table"; + EXPECT_THAT(metadata->ValidateLocation(), IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(ToJson(*metadata), IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(ToJson(*metadata), HasErrorMessage("location must be absolute")); + for (const auto* field : {"next-row-id"}) { + json = MetadataJson(); + json.erase(field); + EXPECT_FALSE(TableMetadataFromJson(json).has_value()); + } +} + +TEST(V4TableMetadataTest, RootManifestRoundTripAndExplicitReaderFailure) { + auto json = MetadataJson(); + json["snapshots"] = nlohmann::json::array({RootSnapshotJson()}); + json["last-sequence-number"] = 1; + json["next-row-id"] = 10; + ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json)); + ASSERT_EQ(metadata->snapshots.size(), 1); + EXPECT_TRUE(metadata->HasV4RootManifest()); + auto snapshot = metadata->snapshots.front(); + EXPECT_EQ(snapshot->root_manifest, "metadata/root.avro"); + EXPECT_TRUE(snapshot->manifest_list.empty()); + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata)); + EXPECT_EQ(serialized["snapshots"][0], RootSnapshotJson()); + EXPECT_FALSE(serialized["snapshots"][0].contains("manifest-list")); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TableMetadataFromJson(serialized)); + EXPECT_EQ(parsed->snapshots.front()->root_manifest, snapshot->root_manifest); + EXPECT_THAT(SnapshotReader(snapshot.get()).Manifests(nullptr), + IsError(ErrorKind::kNotSupported)); +} + +TEST(V4TableMetadataTest, RejectInvalidSnapshotRoots) { + auto json = RootSnapshotJson(); + json["manifest-list"] = "legacy.avro"; + EXPECT_THAT(SnapshotFromJson(json), HasErrorMessage("exactly one")); + for (const auto* field : + {"root-manifest", "sequence-number", "first-row-id", "added-rows", "summary"}) { + json = RootSnapshotJson(); + json.erase(field); + EXPECT_FALSE(SnapshotFromJson(json).has_value()) << field; + } + json = MetadataJson(); + json["format-version"] = 3; + json["snapshots"] = nlohmann::json::array({RootSnapshotJson()}); + EXPECT_THAT(TableMetadataFromJson(json), HasErrorMessage("requires format version 4")); +} + +TEST(V4TableMetadataTest, SnapshotParserUsesTheProvidedFormatVersion) { + for (int8_t version : {1, 2, 3, 4}) { + SCOPED_TRACE(version); + auto root = SnapshotFromJson(RootSnapshotJson(), version); + if (version < TableMetadata::kMinFormatVersionRootManifest) { + EXPECT_THAT(root, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(root, HasErrorMessage("requires format version 4")); + } else { + EXPECT_THAT(root, IsOk()); + } + auto legacy = RootSnapshotJson(); + legacy.erase("root-manifest"); + legacy["manifest-list"] = "legacy.avro"; + EXPECT_THAT(SnapshotFromJson(legacy, version), IsOk()); + } + for (int8_t version : {0, 5}) { + EXPECT_THAT(SnapshotFromJson(RootSnapshotJson(), version), + HasErrorMessage("Unsupported snapshot format version")); + } +} + +TEST(V4TableMetadataTest, SnapshotConstraintStatusesRequireAnArray) { + for (const auto& value : + std::vector{nullptr, nlohmann::json::array()}) { + auto json = RootSnapshotJson(); + json["constraint-statuses"] = value; + EXPECT_THAT(SnapshotFromJson(json), IsOk()); + } + for (const auto& value : + std::vector{true, 0, 0.0, "statuses", nlohmann::json::object()}) { + SCOPED_TRACE(value.dump()); + auto json = RootSnapshotJson(); + json["constraint-statuses"] = value; + auto result = SnapshotFromJson(json); + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("constraint-statuses")); + } + auto json = RootSnapshotJson(); + json["constraint-statuses"] = nlohmann::json::array({{{"constraint-id", 1}}}); + EXPECT_THAT(SnapshotFromJson(json), IsError(ErrorKind::kNotSupported)); +} + +TEST(V4TableMetadataTest, RejectEmptyLegacyManifestListsAcrossMetadataPaths) { + auto legacy_json = RootSnapshotJson(); + legacy_json.erase("root-manifest"); + legacy_json["manifest-list"] = "legacy.avro"; + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(legacy_json)); + snapshot->manifest_list.clear(); + legacy_json["manifest-list"] = ""; + EXPECT_THAT(SnapshotFromJson(legacy_json), IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(SnapshotFromJson(legacy_json), + HasErrorMessage("manifest-list cannot be empty")); + + for (int8_t version : {1, 2, 3, 4}) { + SCOPED_TRACE(version); + EXPECT_THAT(snapshot->ValidateForFormatVersion(version), + HasErrorMessage("manifest-list cannot be empty")); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(version)); + if (version < 4) { + auto builder = TableMetadataBuilder::BuildFrom(metadata.get()); + builder->AddSnapshot(std::make_shared(*snapshot)); + EXPECT_THAT(builder->Build(), HasErrorMessage("manifest-list cannot be empty")); + } + metadata->snapshots.push_back(std::make_shared(*snapshot)); + EXPECT_THAT(ToJson(*metadata), HasErrorMessage("manifest-list cannot be empty")); + } +} + +TEST(V4TableMetadataTest, NewSnapshotsRequireRootsAndMonotonicTimestamps) { + auto json = MetadataJson(); + json["snapshots"] = nlohmann::json::array({RootSnapshotJson()}); + json["last-sequence-number"] = 1; + json["next-row-id"] = 10; + ICEBERG_UNWRAP_OR_FAIL(auto base, TableMetadataFromJson(json)); + auto child_json = RootSnapshotJson(); + child_json["snapshot-id"] = 2; + child_json["parent-snapshot-id"] = 1; + child_json["sequence-number"] = 2; + child_json["first-row-id"] = 10; + for (int64_t timestamp : {999, 1000, 1001}) { + child_json["timestamp-ms"] = timestamp; + ICEBERG_UNWRAP_OR_FAIL(auto child, SnapshotFromJson(child_json)); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->AddSnapshot(std::move(child)); + if (timestamp <= 1000) { + EXPECT_THAT(builder->Build(), IsError(ErrorKind::kValidationFailed)); + } else { + ICEBERG_UNWRAP_OR_FAIL(auto updated, builder->Build()); + EXPECT_EQ(updated->next_row_id, 20); + } + } + child_json.erase("root-manifest"); + child_json["manifest-list"] = "legacy.avro"; + child_json["timestamp-ms"] = 1001; + ICEBERG_UNWRAP_OR_FAIL(auto child, SnapshotFromJson(child_json)); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->AddSnapshot(std::move(child)); + EXPECT_THAT(builder->Build(), + HasErrorMessage("New v4 snapshots require root-manifest")); +} + +TEST(V4TableMetadataTest, StaleSequenceIsRetryableBeforeParentTimestampValidation) { + auto json = MetadataJson(); + json["snapshots"] = nlohmann::json::array({RootSnapshotJson()}); + json["last-sequence-number"] = 1; + json["next-row-id"] = 10; + ICEBERG_UNWRAP_OR_FAIL(auto base, TableMetadataFromJson(json)); + auto child_json = RootSnapshotJson(); + child_json["snapshot-id"] = 2; + child_json["parent-snapshot-id"] = 1; + child_json["first-row-id"] = 10; + // Both timestamps are invalid; a stale sequence must take retry precedence. + for (int64_t timestamp : {999, 1000}) { + child_json["timestamp-ms"] = timestamp; + for (int64_t sequence : {1, 2}) { + child_json["sequence-number"] = sequence; + ICEBERG_UNWRAP_OR_FAIL(auto child, SnapshotFromJson(child_json, 4)); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->AddSnapshot(std::move(child)); + auto result = builder->Build(); + EXPECT_THAT(result, IsError(sequence == 1 ? ErrorKind::kRetryableValidationFailed + : ErrorKind::kValidationFailed)); + EXPECT_THAT(result, + HasErrorMessage(sequence == 1 ? "sequence number" : "timestamp")); + EXPECT_TRUE(builder->changes().empty()); + } + } +} + +TEST(V4TableMetadataTest, RejectsWritingIdentifierFieldsWithoutMigration) { + for (bool historical_schema : {false, true}) { + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata()); + ICEBERG_UNWRAP_OR_FAIL( + auto schema, Schema::Make({SchemaField::MakeRequired(1, "id", int64())}, + historical_schema ? 1 : 0, std::vector{1})); + if (historical_schema) { + metadata->schemas.push_back(std::move(schema)); + } else { + metadata->schemas[0] = std::move(schema); + } + EXPECT_THAT(ToJson(*metadata), IsError(ErrorKind::kNotSupported)); + auto builder = TableMetadataBuilder::BuildFrom(metadata.get()); + EXPECT_THAT(builder->Build(), IsError(ErrorKind::kNotSupported)); + ASSERT_EQ(metadata->schemas.back()->IdentifierFieldIds(), std::vector{1}); + } +} + +TEST(V4TableMetadataTest, IgnoreLegacyIdentifiersAndRejectUnimplementedConstraints) { + auto json = MetadataJson(); + json["schemas"][0]["identifier-field-ids"] = {999}; + ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json)); + ICEBERG_UNWRAP_OR_FAIL(auto schema, metadata->Schema()); + EXPECT_TRUE(schema->IdentifierFieldIds().empty()); + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata)); + EXPECT_FALSE(serialized["schemas"][0].contains("identifier-field-ids")); + for (const auto* field : {"constraints", "last-constraint-id"}) { + json = MetadataJson(); + json[field] = field == std::string_view("constraints") + ? nlohmann::json::array({{{"constraint-id", 1}}}) + : nlohmann::json(1); + EXPECT_THAT(TableMetadataFromJson(json), IsError(ErrorKind::kNotSupported)); + } + json = MetadataJson(); + json["format-version"] = 3; + json["schemas"][0]["identifier-field-ids"] = {1}; + ICEBERG_UNWRAP_OR_FAIL(auto base, TableMetadataFromJson(json)); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->UpgradeFormatVersion(4); + EXPECT_THAT(builder->Build(), HasErrorMessage("primary-key constraint migration")); + for (const auto& malformed_schema : std::vector{nullptr, 1, "schema"}) { + json = MetadataJson(); + json["schemas"] = nlohmann::json::array({malformed_schema}); + EXPECT_FALSE(TableMetadataFromJson(json).has_value()); + } +} + +TEST(V4TableMetadataTest, ConstraintFieldsRequireTheirDeclaredJsonTypes) { + auto json = MetadataJson(); + json["constraints"] = nlohmann::json::array(); + json["last-constraint-id"] = 0; + EXPECT_THAT(TableMetadataFromJson(json), IsOk()); + for (const auto* field : {"constraints", "last-constraint-id"}) { + auto nullable = json; + nullable[field] = nullptr; + EXPECT_THAT(TableMetadataFromJson(nullable), IsOk()); + } + json["constraints"] = nullptr; + json["last-constraint-id"] = nullptr; + EXPECT_THAT(TableMetadataFromJson(json), IsOk()); + + for (const auto& value : std::vector{"constraints", true, 0, 0.0, + nlohmann::json::object()}) { + SCOPED_TRACE(value.dump()); + json = MetadataJson(); + json["constraints"] = value; + auto result = TableMetadataFromJson(json); + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("constraints")); + } + for (const auto& value : std::vector{ + "0", true, 0.0, 1.5, 2147483648LL, -2147483649LL, 18446744073709551615ULL, + nlohmann::json::array(), nlohmann::json::object()}) { + SCOPED_TRACE(value.dump()); + json = MetadataJson(); + json["last-constraint-id"] = value; + auto result = TableMetadataFromJson(json); + EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError)); + EXPECT_THAT(result, HasErrorMessage("last-constraint-id")); + } +} + +std::shared_ptr MemoryIO() { + auto io = std::make_shared<::testing::NiceMock>(); + ON_CALL(*io, ReadFile) + .WillByDefault( + [ptr = io.get()](const std::string& path, std::optional size) { + return ptr->FileIO::ReadFile(path, size); + }); + ON_CALL(*io, WriteFile) + .WillByDefault([ptr = io.get()](const std::string& path, std::string_view data) { + return ptr->FileIO::WriteFile(path, data); + }); + return io; +} + +TEST(V4TableMetadataTest, CatalogCreateLoadAndPropertyUpgrade) { + auto io = MemoryIO(); + ICEBERG_UNWRAP_OR_FAIL(auto catalog, + InMemoryCatalog::Make("v4", io, "s3://warehouse", {})); + for (const auto* initial_version : {"2", "4"}) { + TableIdentifier ident{.name = std::string("version_") + initial_version}; + ICEBERG_UNWRAP_OR_FAIL( + auto table, + catalog->CreateTable(ident, TestSchema(), PartitionSpec::Unpartitioned(), + SortOrder::Unsorted(), "s3://warehouse/" + ident.name, + {{"format-version", initial_version}})); + ICEBERG_UNWRAP_OR_FAIL(auto update, table->NewUpdateProperties()); + update->Set("format-version", "4"); + ASSERT_THAT(update->Commit(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog->LoadTable(ident)); + EXPECT_EQ(loaded->metadata()->format_version, 4); + EXPECT_FALSE(loaded->metadata()->properties.configs().contains("format-version")); + ICEBERG_UNWRAP_OR_FAIL(auto append, loaded->NewFastAppend()); + EXPECT_THAT(append->Commit(), + HasErrorMessage("Writing v4 manifest trees is not supported")); + } +} + +TEST(V4TableMetadataTest, StagedCatalogCreate) { + auto io = MemoryIO(); + ICEBERG_UNWRAP_OR_FAIL(auto catalog, + InMemoryCatalog::Make("v4", io, "s3://warehouse", {})); + TableIdentifier ident{.name = "staged_v4"}; + ICEBERG_UNWRAP_OR_FAIL( + auto transaction, + catalog->StageCreateTable(ident, TestSchema(), PartitionSpec::Unpartitioned(), + SortOrder::Unsorted(), "s3://warehouse/staged_v4", + {{"format-version", "4"}})); + ASSERT_THAT(transaction->Commit(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog->LoadTable(ident)); + EXPECT_EQ(loaded->metadata()->format_version, 4); +} + +TEST(V4TableMetadataTest, RejectInvalidVersionPropertiesAndIncompleteSnapshotWrites) { + for (const auto* version : {"0", "-252", "5", "260", "invalid"}) { + EXPECT_FALSE(TableMetadata::Make(*TestSchema(), *PartitionSpec::Unpartitioned(), + *SortOrder::Unsorted(), "s3://bucket/table", + {{"format-version", version}}) + .has_value()); + } + auto json = RootSnapshotJson(); + json["summary"] = nlohmann::json::object(); + EXPECT_FALSE(SnapshotFromJson(json).has_value()); + json = RootSnapshotJson(); + json["sequence-number"] = 0; + EXPECT_FALSE(SnapshotFromJson(json).has_value()); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(RootSnapshotJson())); + snapshot->first_row_id.reset(); + metadata->snapshots.push_back(std::move(snapshot)); + EXPECT_THAT(ToJson(*metadata), HasErrorMessage("first-row-id")); +} + +TEST(V4TableMetadataTest, MissingOrNullUuidUsesTheDefaultForEveryVersion) { + for (int version : {1, 2, 3, 4}) { + auto json = MetadataJson(); + json["format-version"] = version; + for (bool explicit_null : {false, true}) { + if (explicit_null) { + json["table-uuid"] = nullptr; + } else { + json.erase("table-uuid"); + } + ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json)); + EXPECT_TRUE(metadata->table_uuid.empty()); + } + } +} + +TEST(V4TableMetadataTest, NullLocationIsAbsentAndCannotProduceDefaultDataPaths) { + auto json = MetadataJson(); + json["location"] = nullptr; + ICEBERG_UNWRAP_OR_FAIL(auto metadata, TableMetadataFromJson(json)); + EXPECT_TRUE(metadata->location.empty()); + ICEBERG_UNWRAP_OR_FAIL(auto serialized, ToJson(*metadata)); + EXPECT_FALSE(serialized.contains("location")); + ICEBERG_UNWRAP_OR_FAIL( + auto table, Table::Make(TableIdentifier{.name = "unlocated"}, std::move(metadata), + "s3://bucket/metadata.json", MemoryIO(), + std::make_shared<::testing::NiceMock>())); + EXPECT_THAT(table->location_provider(), IsError(ErrorKind::kInvalidArgument)); + for (int version : {1, 2, 3}) { + json["format-version"] = version; + EXPECT_FALSE(TableMetadataFromJson(json).has_value()); + } +} + +TEST(V4TableMetadataTest, MissingTableLocationCanUseAnExplicitMetadataPath) { + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata()); + metadata->location.clear(); + auto io = MemoryIO(); + auto missing = TableMetadataUtil::Write(*io, nullptr, "", *metadata); + EXPECT_THAT(missing, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(missing, HasErrorMessage(TableProperties::kWriteMetadataLocation.key())); + metadata->properties.Set(TableProperties::kWriteMetadataLocation, + std::string("s3://bucket/explicit-metadata")); + ICEBERG_UNWRAP_OR_FAIL(auto path, + TableMetadataUtil::Write(*io, nullptr, "", *metadata)); + EXPECT_THAT(path, ::testing::StartsWith("s3://bucket/explicit-metadata/")); +} + +TEST(V4TableMetadataTest, UpgradeBeforeSelectingCurrentSchema) { + ICEBERG_UNWRAP_OR_FAIL(auto metadata, NewMetadata(3)); + metadata->current_schema_id = -1; + auto builder = TableMetadataBuilder::BuildFrom(metadata.get()); + builder->AddSchema(TestSchema()).UpgradeFormatVersion(4).SetCurrentSchema(0); + ICEBERG_UNWRAP_OR_FAIL(auto upgraded, builder->Build()); + EXPECT_EQ(upgraded->format_version, 4); + EXPECT_EQ(upgraded->current_schema_id, 0); +} + +TEST(V4TableMetadataTest, RootOperationValidationIsSharedAcrossMetadataPaths) { + for (const auto& operation : + std::vector>{std::nullopt, "", "invalid"}) { + auto json = RootSnapshotJson(); + if (operation.has_value()) { + json["summary"]["operation"] = operation.value(); + } else { + json["summary"].erase("operation"); + } + EXPECT_THAT(SnapshotFromJson(json), HasErrorMessage("valid operation")); + + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(RootSnapshotJson())); + if (operation.has_value()) { + snapshot->summary["operation"] = operation.value(); + } else { + snapshot->summary.erase("operation"); + } + EXPECT_THAT(snapshot->ValidateForFormatVersion(4), + IsError(ErrorKind::kValidationFailed)); + ICEBERG_UNWRAP_OR_FAIL(auto base, NewMetadata()); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->AddSnapshot(std::make_shared(*snapshot)); + EXPECT_THAT(builder->Build(), HasErrorMessage("valid operation")); + base->snapshots.push_back(std::move(snapshot)); + EXPECT_THAT(ToJson(*base), HasErrorMessage("valid operation")); + } +} + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/test/v4_types_test.cc b/src/iceberg/test/v4_types_test.cc new file mode 100644 index 000000000..82adce9d3 --- /dev/null +++ b/src/iceberg/test/v4_types_test.cc @@ -0,0 +1,185 @@ +/* + * 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 "iceberg/manifest/v4_types.h" + +#include +#include +#include +#include + +#include +#include +#include + +#include "iceberg/json_serde_internal.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/test/matchers.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +// Independent wire-schema fixtures from the v4 manifest schema in the spec. +// Check IDs, names, order, types, and requiredness through the schema serializer. +void CheckSchema(const StructType& type, const nlohmann::json& expected) { + ICEBERG_UNWRAP_OR_FAIL(auto actual, ToJson(type)); + EXPECT_EQ(actual, expected); + ICEBERG_UNWRAP_OR_FAIL(auto parsed, TypeFromJson(expected)); + EXPECT_EQ(type, *parsed); +} + +TEST(V4TypesTest, FileTrackingSchema) { + const auto expected = nlohmann::json::parse(R"({ + "type": "struct", + "fields": [ + {"id": 0, "name": "status", "type": "int", "required": true}, + {"id": 1, "name": "snapshot_id", "type": "long", "required": false}, + {"id": 5, "name": "modified_snapshot_id", "type": "long", "required": false}, + {"id": 3, "name": "sequence_number", "type": "long", "required": false}, + {"id": 4, "name": "file_sequence_number", "type": "long", "required": false}, + {"id": 142, "name": "first_row_id", "type": "long", "required": false}, + {"id": 6, "name": "deleted_positions", "type": "binary", "required": false}, + {"id": 7, "name": "replaced_positions", "type": "binary", "required": false} + ] + })"); + auto type = FileTracking::Type(); + EXPECT_EQ(type, FileTracking::Type()); + CheckSchema(*type, expected); +} + +TEST(V4TypesTest, DeletionVectorMetadataSchema) { + const auto expected = nlohmann::json::parse(R"({ + "type": "struct", + "fields": [ + {"id": 155, "name": "location", "type": "string", "required": true}, + {"id": 144, "name": "offset", "type": "long", "required": true}, + {"id": 145, "name": "size_in_bytes", "type": "long", "required": true}, + {"id": 156, "name": "cardinality", "type": "long", "required": true}, + {"id": 149, "name": "key_metadata", "type": "binary", "required": false} + ] + })"); + auto type = DeletionVectorMetadata::Type(); + EXPECT_EQ(type, DeletionVectorMetadata::Type()); + CheckSchema(*type, expected); +} + +TEST(V4TypesTest, ManifestInfoSchema) { + const auto expected = nlohmann::json::parse(R"({ + "type": "struct", + "fields": [ + {"id": 521, "name": "format_version", "type": "int", "required": true}, + {"id": 522, "name": "dv", "type": "binary", "required": false}, + {"id": 504, "name": "added_files_count", "type": "int", "required": true}, + {"id": 505, "name": "existing_files_count", "type": "int", "required": true}, + {"id": 525, "name": "modified_files_count", "type": "int", "required": true}, + {"id": 506, "name": "deleted_files_count", "type": "int", "required": true}, + {"id": 523, "name": "replaced_files_count", "type": "int", "required": true}, + {"id": 512, "name": "added_rows_count", "type": "long", "required": true}, + {"id": 513, "name": "existing_rows_count", "type": "long", "required": true}, + {"id": 526, "name": "modified_rows_count", "type": "long", "required": true}, + {"id": 514, "name": "deleted_rows_count", "type": "long", "required": true}, + {"id": 524, "name": "replaced_rows_count", "type": "long", "required": true}, + {"id": 516, "name": "min_sequence_number", "type": "long", "required": true} + ] + })"); + auto type = ManifestInfo::Type(); + EXPECT_EQ(type, ManifestInfo::Type()); + CheckSchema(*type, expected); +} + +struct TrackingStatusCase { + int32_t value; + FileTrackingStatus status; + bool alive; +}; + +class FileTrackingStatusTest : public ::testing::TestWithParam {}; + +TEST_P(FileTrackingStatusTest, DecodeAndPlanLiveEntries) { + const auto& test_case = GetParam(); + ICEBERG_UNWRAP_OR_FAIL(auto status, FileTrackingStatusFromInt(test_case.value)); + EXPECT_EQ(status, test_case.status); + EXPECT_EQ(static_cast(status), test_case.value); + EXPECT_EQ(FileTracking{.status = status}.IsAlive(), test_case.alive); +} + +INSTANTIATE_TEST_SUITE_P( + AllStatuses, FileTrackingStatusTest, + ::testing::Values(TrackingStatusCase{0, FileTrackingStatus::kExisting, true}, + TrackingStatusCase{1, FileTrackingStatus::kAdded, true}, + TrackingStatusCase{2, FileTrackingStatus::kDeleted, false}, + TrackingStatusCase{3, FileTrackingStatus::kReplaced, false}, + TrackingStatusCase{4, FileTrackingStatus::kModified, true})); + +TEST(V4TypesTest, InvalidTrackingStatuses) { + for (int32_t value : {-1, 5, std::numeric_limits::min(), + std::numeric_limits::max()}) { + SCOPED_TRACE(value); + EXPECT_THAT(FileTrackingStatusFromInt(value), IsError(ErrorKind::kInvalidArgument)); + } +} + +TEST(V4TypesTest, LegacyStatusDecoderRejectsV4Statuses) { + for (int32_t value : {0, 1, 2}) { + EXPECT_THAT(ManifestStatusFromInt(value), IsOk()); + } + for (int32_t value : {3, 4}) { + EXPECT_THAT(ManifestStatusFromInt(value), IsError(ErrorKind::kInvalidArgument)); + } +} + +TEST(V4TypesTest, TrackingPreservesMissingInheritanceFieldsAndBitmaps) { + FileTracking tracking; + EXPECT_FALSE(tracking.snapshot_id.has_value()); + EXPECT_FALSE(tracking.modified_snapshot_id.has_value()); + EXPECT_FALSE(tracking.sequence_number.has_value()); + EXPECT_FALSE(tracking.file_sequence_number.has_value()); + EXPECT_FALSE(tracking.first_row_id.has_value()); + EXPECT_FALSE(tracking.deleted_positions.has_value()); + EXPECT_FALSE(tracking.replaced_positions.has_value()); + + // Missing sequence numbers must remain distinguishable from sequence number 0. + tracking.sequence_number = 0; + EXPECT_TRUE(tracking.sequence_number.has_value()); + EXPECT_EQ(tracking.sequence_number.value(), 0); + + tracking.deleted_positions = std::vector{}; + const auto copy = tracking; + ASSERT_TRUE(copy.deleted_positions.has_value()); + EXPECT_TRUE(copy.deleted_positions->empty()); + EXPECT_FALSE(copy.replaced_positions.has_value()); +} + +TEST(V4TypesTest, OptionalBinaryMetadataPreservesAbsence) { + DeletionVectorMetadata dv; + ManifestInfo manifest; + EXPECT_FALSE(dv.key_metadata.has_value()); + EXPECT_FALSE(manifest.dv.has_value()); + dv.key_metadata = std::vector{0, 255}; + manifest.dv = std::vector{}; + const auto dv_copy = dv; + const auto manifest_copy = manifest; + EXPECT_EQ(dv_copy.key_metadata, dv.key_metadata); + ASSERT_TRUE(manifest_copy.dv.has_value()); + EXPECT_TRUE(manifest_copy.dv->empty()); +} + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc index fd14308e8..fee405134 100644 --- a/src/iceberg/update/expire_snapshots.cc +++ b/src/iceberg/update/expire_snapshots.cc @@ -966,7 +966,19 @@ Result ExpireSnapshots::Apply() { } }); + if (cleanup_level_ != CleanupLevel::kNone && !result.snapshot_ids_to_remove.empty() && + base.HasV4RootManifest()) { + return NotSupported( + "Snapshot expiration file cleanup with v4 root manifests " + "is not supported yet; use CleanupLevel::kNone"); + } + if (clean_expired_metadata_) { + if (base.HasV4RootManifest()) { + return NotSupported( + "Cleaning expired schemas and partition specs with v4 root manifests " + "is not supported yet; disable CleanExpiredMetadata to expire snapshots"); + } ICEBERG_ASSIGN_OR_RAISE( auto reachable_ids, ParallelCollect( @@ -1026,6 +1038,17 @@ Status ExpireSnapshots::Finalize(const TableMetadata& metadata_after_expiration) const TableMetadata& metadata_before_expiration = *apply_result->metadata_before_expiration; + // Later transaction updates or the catalog's committed result may introduce + // root manifests after Apply(). Never treat untraversable roots as empty. + // The commit has succeeded, so retain files and signal the skipped cleanup. + if (metadata_before_expiration.HasV4RootManifest() || + metadata_after_expiration.HasV4RootManifest()) { + ICEBERG_LOG_WARN( + "Skipped snapshot expiration file cleanup: v4 root manifests prevent " + "reachability checks; expired snapshot files were retained"); + return {}; + } + // Pick incremental cleanup when the expiration is a simple linear-ancestry walk: // no explicit snapshot IDs, no removed snapshots outside main ancestry, and no // retained snapshots outside main ancestry. diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc index d6a1066ac..f2a795f4b 100644 --- a/src/iceberg/update/snapshot_update.cc +++ b/src/iceberg/update/snapshot_update.cc @@ -311,6 +311,9 @@ int64_t SnapshotUpdate::SnapshotId() { Result SnapshotUpdate::Apply() { commit_metrics_->attempts->Increment(); ICEBERG_RETURN_UNEXPECTED(CheckErrors()); + if (base().format_version >= TableMetadata::kMinFormatVersionRootManifest) { + return NotSupported("Writing v4 manifest trees is not supported yet"); + } ICEBERG_ASSIGN_OR_RAISE(auto parent_snapshot, SnapshotUtil::OptionalLatestSnapshot(base(), target_branch_)); diff --git a/src/iceberg/update/update_properties.cc b/src/iceberg/update/update_properties.cc index c97175569..b42a8806d 100644 --- a/src/iceberg/update/update_properties.cc +++ b/src/iceberg/update/update_properties.cc @@ -88,7 +88,8 @@ Result UpdateProperties::Apply() { ICEBERG_ASSIGN_OR_RAISE(auto parsed_version, StringUtils::ParseNumber(iter->second)); - if (parsed_version > TableMetadata::kSupportedTableFormatVersion) { + if (parsed_version < 1 || + parsed_version > TableMetadata::kSupportedTableFormatVersion) { return InvalidArgument( "Cannot upgrade table to unsupported format version: v{} (supported: v{})", parsed_version, TableMetadata::kSupportedTableFormatVersion);