Fix antalya-26.6 build: adapt Iceberg TRUNCATE to 26.6 writer API (#2125)

The #1655 backport was written against the antalya-26.3 Iceberg writer API
and did not compile on antalya-26.6, breaking Build (arm_tidy) and Fast test:

* Constant.h re-added the `deleted_records` / `deleted_data_files` field
  aliases that antalya-26.6 already defines, causing macro redefinition
  errors. Drop the duplicate definitions.

* IcebergMetadata::truncate() used the 26.3 FileNamesGenerator /
  MetadataGenerator / generateManifestList signatures. Rewrite it to follow
  the antalya-26.6 writer (IcebergStorageSink in IcebergWrites.cpp):
  - 4-arg FileNamesGenerator over resolver.getTableLocation();
  - generateMetadataPathWithInfo() instead of the removed generateMetadataName();
  - generateNextMetadata() now returns {snapshot, manifest_list_path}; resolve
    storage paths via IcebergPathResolver::resolve() and the catalog path via
    resolveForCatalog();
  - generateManifestList(resolver, ...) with empty entry/size vectors.

Metadata-only truncate semantics are unchanged. Not built locally; validated
by CI on this PR.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: blau-ai

src/Storages/ObjectStorage/DataLakes/Iceberg/Constant.h

@@ -123,9 +123,7 @@ DEFINE_ICEBERG_FIELD_ALIAS(partition_spec, partition-spec);
 DEFINE_ICEBERG_FIELD_ALIAS(partition_specs, partition-specs);
 DEFINE_ICEBERG_FIELD_ALIAS(spec_id, spec-id);
 DEFINE_ICEBERG_FIELD_ALIAS(added_records, added-records);
-DEFINE_ICEBERG_FIELD_ALIAS(deleted_records, deleted-records);
 DEFINE_ICEBERG_FIELD_ALIAS(added_data_files, added-data-files);
-DEFINE_ICEBERG_FIELD_ALIAS(deleted_data_files, deleted-data-files);
 DEFINE_ICEBERG_FIELD_ALIAS(added_delete_files, added-delete-files);
 DEFINE_ICEBERG_FIELD_ALIAS(added_position_delete_files, added-position-delete-files);
 DEFINE_ICEBERG_FIELD_ALIAS(added_position_deletes, added-position-deletes);

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -635,70 +635,54 @@ void IcebergMetadata::truncate(ContextPtr context, std::shared_ptr<DataLake::ICa
     // (distinct from snapshot ID 0 which is a valid snapshot).
     Int64 parent_snapshot_id = actual_table_state_snapshot.snapshot_id.value_or(-1);
 
-    auto config_path = persistent_components.table_path;
-    if (!config_path.starts_with('/')) config_path = '/' + config_path;
-    if (!config_path.ends_with('/')) config_path += "/";
+    // On antalya-26.6 all metadata paths flow through the table's IcebergPathResolver:
+    // FileNamesGenerator produces IcebergPathFromMetadata values (relative to the table
+    // location), and resolver.resolve() / resolveForCatalog() turn those into the storage
+    // path for I/O and the fully-qualified path the catalog expects. This mirrors the
+    // write path in IcebergStorageSink (see IcebergWrites.cpp) so transactional (REST) and
+    // non-transactional catalogs are handled uniformly.
+    const auto & resolver = persistent_components.path_resolver;
 
     bool is_transactional = (catalog != nullptr && catalog->isTransactional());
 
-    // Transactional catalogs (e.g. REST) require a fully-qualified blob URI
-    // (scheme://bucket/path) so the catalog can resolve the metadata location
-    // independently of any local path configuration. Non-transactional catalogs
-    // use bare paths relative to the object storage root.
-    FileNamesGenerator filename_generator;
-    if (is_transactional || context->getSettingsRef()[Setting::write_full_path_in_iceberg_metadata])
-    {
-        String location = metadata_object->getValue<String>(Iceberg::f_location);
-        if (!location.ends_with("/")) location += "/";
-        filename_generator = FileNamesGenerator(
-            location, config_path, is_transactional,
-            persistent_components.metadata_compression_method, write_format);
-    }
-    else
-    {
-        filename_generator = FileNamesGenerator(
-            config_path, config_path, false,
-            persistent_components.metadata_compression_method, write_format);
-    }
+    FileNamesGenerator filename_generator(
+        resolver.getTableLocation(),
+        is_transactional,
+        persistent_components.metadata_compression_method,
+        write_format);
 
     Int32 new_metadata_version = actual_table_state_snapshot.metadata_version + 1;
     filename_generator.setVersion(new_metadata_version);
 
-    auto [metadata_name, storage_metadata_name] = filename_generator.generateMetadataName();
+    auto metadata_info = filename_generator.generateMetadataPathWithInfo();
 
-    auto [new_snapshot, manifest_list_name, storage_manifest_list_name] = MetadataGenerator(metadata_object).generateNextMetadata(
-        filename_generator, metadata_name, parent_snapshot_id,
+    auto [new_snapshot, manifest_list_path] = MetadataGenerator(metadata_object).generateNextMetadata(
+        filename_generator, metadata_info.path, parent_snapshot_id,
         /* added_files */ 0, /* added_records */ 0, /* added_files_size */ 0,
         /* num_partitions */ 0, /* added_delete_files */ 0, /* num_deleted_rows */ 0,
         std::nullopt, std::nullopt, /*is_truncate=*/true);
 
+    auto storage_manifest_list_name = resolver.resolve(manifest_list_path);
+
     auto write_settings = context->getWriteSettings();
     auto buf = object_storage->writeObject(
         StoredObject(storage_manifest_list_name),
         WriteMode::Rewrite, std::nullopt,
         DBMS_DEFAULT_BUFFER_SIZE, write_settings);
 
-    generateManifestList(filename_generator, metadata_object, object_storage,
-        context, {}, new_snapshot, 0, *buf, Iceberg::FileContentType::DATA, /*use_previous_snapshots=*/false);
+    // Truncate writes a metadata-only overwrite snapshot: an empty manifest list
+    // (no manifest entries, no sizes) that supersedes all previous snapshots.
+    generateManifestList(resolver, metadata_object, object_storage,
+        context, {}, new_snapshot, {}, *buf, Iceberg::FileContentType::DATA, /*use_previous_snapshots=*/false);
     buf->finalize();
 
     String metadata_content = dumpMetadataObjectToString(metadata_object);
-    writeMessageToFile(metadata_content, storage_metadata_name, object_storage,
+    writeMessageToFile(metadata_content, resolver.resolve(metadata_info.path), object_storage,
         context, "*", "", persistent_components.metadata_compression_method);
 
     if (catalog)
     {
-        // Transactional catalogs require a fully-qualified blob URI so the catalog
-        // can resolve the metadata location independently of local path configuration.
-        String catalog_filename = metadata_name;
-        if (is_transactional)
-        {
-            // Build full URI from the table's location field (e.g. "s3://bucket/namespace.table")
-            // combined with the relative metadata name.
-            String location = metadata_object->getValue<String>(Iceberg::f_location);
-            if (!location.ends_with("/")) location += "/";
-            catalog_filename = location + metadata_name;
-        }
+        String catalog_filename = resolver.resolveForCatalog(metadata_info.path);
 
         const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName());
         if (!catalog->updateMetadata(namespace_name, table_name, catalog_filename, new_snapshot))