some small adaptations

Author: arthurpassos

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

@@ -1492,7 +1492,7 @@ bool IcebergMetadata::commitImportPartitionTransactionImpl(
 
     String manifest_entry_name;
     String storage_manifest_entry_name;
-    Int32 manifest_lengths = 0;
+    Int64 manifest_lengths = 0;
 
     /// Tracks whether the snapshot has become visible to readers.
     /// For the file-based layout that happens as soon as writeMetadataFileAndVersionHint

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

@@ -1374,6 +1374,11 @@ IcebergImportSink::IcebergImportSink(
         new_file_path_callback);
 }
 
+IcebergImportSink::~IcebergImportSink()
+{
+    cancelBuffers();
+}
+
 void IcebergImportSink::consume(Chunk & chunk)
 {
     if (isCancelled())
@@ -1385,7 +1390,10 @@ void IcebergImportSink::consume(Chunk & chunk)
 void IcebergImportSink::onFinish()
 {
     if (isCancelled())
+    {
+        cancelBuffers();
         return;
+    }
 
     finalizeBuffers();
 
@@ -1408,6 +1416,11 @@ void IcebergImportSink::onFinish()
     releaseBuffers();
 }
 
+void IcebergImportSink::onException(std::exception_ptr /* exception */)
+{
+    cancelBuffers();
+}
+
 void IcebergImportSink::finalizeBuffers()
 {
     writer->finalize();

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

@@ -182,13 +182,14 @@ class IcebergImportSink : public SinkToStorage
         const DataLakeStorageSettings & data_lake_settings_,
         std::function<void(const std::string &)> new_file_path_callback_ = {});
 
-    ~IcebergImportSink() override = default;
+    ~IcebergImportSink() override;
 
     String getName() const override { return "IcebergImportSink"; }
 
     void consume(Chunk & chunk) override;
 
     void onFinish() override;
+    void onException(std::exception_ptr exception) override;
 
 private:
     void finalizeBuffers();