fix: address avro blob and executor review issues (alibaba#378)

Author: zjw1111

src/paimon/common/data/blob_defs.h

@@ -53,8 +53,8 @@ class BlobDefs {
     static constexpr int32_t kContentStartOffset = 4;
     /// Total metadata length per bin: magic(4) + bin_length(8) + crc32(4) = 16.
     static constexpr int32_t kTotalMetaLength = 16;
-    /// Blob file header length: index_len(4) + version(1) = 5.
-    static constexpr uint32_t kBlobFileHeaderLength = 5;
+    /// Blob file footer length: index_len(4) + version(1) = 5.
+    static constexpr uint32_t kBlobFileFooterLength = 5;
 };
 
 }  // namespace paimon

src/paimon/format/avro/avro_direct_decoder.h

@@ -33,6 +33,9 @@ namespace paimon::avro {
 
 class AvroDirectDecoder {
  public:
+    AvroDirectDecoder() = delete;
+    ~AvroDirectDecoder() = delete;
+
     /// Context for reusing scratch buffers during Avro decoding
     ///
     /// Avoids frequent small allocations by reusing temporary buffers across multiple decode

src/paimon/format/avro/avro_direct_encoder.cpp

@@ -354,11 +354,11 @@ Status AvroDirectEncoder::EncodeArrowToAvro(const ::avro::NodePtr& avro_node,
                     return Status::Invalid(fmt::format("AVRO_MAP keys must be StringArray, got {}",
                                                        keys->type()->ToString()));
                 }
+                const auto& string_array =
+                    arrow::internal::checked_cast<const arrow::StringArray&>(*keys);
 
                 for (int64_t i = start; i < end; ++i) {
                     encoder->startItem();
-                    const auto& string_array =
-                        arrow::internal::checked_cast<const arrow::StringArray&>(*keys);
                     std::string_view key_value = string_array.GetView(i);
                     encoder->encodeString(std::string(key_value));
 

src/paimon/format/avro/avro_direct_encoder.h

@@ -33,6 +33,9 @@ namespace paimon::avro {
 
 class AvroDirectEncoder {
  public:
+    AvroDirectEncoder() = delete;
+    ~AvroDirectEncoder() = delete;
+
     /// Context for reusing scratch buffers during Avro encoding
     ///
     /// Avoids frequent small allocations by reusing temporary buffers across multiple encode

src/paimon/format/avro/avro_file_format.cpp

@@ -39,7 +39,7 @@ AvroFileFormat::AvroFileFormat(const std::map<std::string, std::string>& options
 
 Result<std::unique_ptr<ReaderBuilder>> AvroFileFormat::CreateReaderBuilder(
     int32_t batch_size) const {
-    return std::make_unique<AvroReaderBuilder>(options_, batch_size);
+    return std::make_unique<AvroReaderBuilder>(batch_size);
 }
 
 Result<std::unique_ptr<WriterBuilder>> AvroFileFormat::CreateWriterBuilder(

src/paimon/format/avro/avro_reader_builder.h

@@ -32,8 +32,8 @@ namespace paimon::avro {
 
 class AvroReaderBuilder : public ReaderBuilder {
  public:
-    AvroReaderBuilder(const std::map<std::string, std::string>& options, int32_t batch_size)
-        : batch_size_(batch_size), pool_(GetDefaultPool()), options_(options) {}
+    explicit AvroReaderBuilder(int32_t batch_size)
+        : batch_size_(batch_size), pool_(GetDefaultPool()) {}
 
     ReaderBuilder* WithMemoryPool(const std::shared_ptr<MemoryPool>& pool) override {
         pool_ = pool;
@@ -52,7 +52,6 @@ class AvroReaderBuilder : public ReaderBuilder {
  private:
     const int32_t batch_size_;
     std::shared_ptr<MemoryPool> pool_;
-    const std::map<std::string, std::string> options_;
 };
 
 }  // namespace paimon::avro

src/paimon/format/blob/blob_file_batch_reader.cpp

@@ -17,7 +17,6 @@
 #include "paimon/format/blob/blob_file_batch_reader.h"
 
 #include <algorithm>
-#include <future>
 #include <numeric>
 
 #include "arrow/api.h"
@@ -27,7 +26,6 @@
 #include "arrow/util/bit_util.h"
 #include "fmt/format.h"
 #include "paimon/common/data/blob_utils.h"
-#include "paimon/common/executor/future.h"
 #include "paimon/common/io/offset_input_stream.h"
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
@@ -52,24 +50,24 @@ Result<std::unique_ptr<BlobFileBatchReader>> BlobFileBatchReader::Create(
 
     PAIMON_ASSIGN_OR_RAISE(int64_t file_size, input_stream->Length());
     PAIMON_RETURN_NOT_OK(
-        input_stream->Seek(file_size - BlobDefs::kBlobFileHeaderLength, FS_SEEK_SET));
-    int8_t header[BlobDefs::kBlobFileHeaderLength];
+        input_stream->Seek(file_size - BlobDefs::kBlobFileFooterLength, FS_SEEK_SET));
+    int8_t footer[BlobDefs::kBlobFileFooterLength];
     PAIMON_ASSIGN_OR_RAISE(
         int64_t actual_size,
-        input_stream->Read(reinterpret_cast<char*>(header), BlobDefs::kBlobFileHeaderLength));
-    if (actual_size != BlobDefs::kBlobFileHeaderLength) {
+        input_stream->Read(reinterpret_cast<char*>(footer), BlobDefs::kBlobFileFooterLength));
+    if (actual_size != BlobDefs::kBlobFileFooterLength) {
         return Status::Invalid(
-            fmt::format("actual read size {} not match with expect header length {}", actual_size,
-                        BlobDefs::kBlobFileHeaderLength));
+            fmt::format("actual read size {} not match with expect footer length {}", actual_size,
+                        BlobDefs::kBlobFileFooterLength));
     }
-    int8_t version = header[4];
+    int8_t version = footer[4];
     if (version != BlobDefs::kFileVersion) {
         return Status::Invalid(fmt::format(
             "create blob format reader failed. unsupported blob file version: {}", version));
     }
-    int32_t index_length = GetIndexLength(header, 0);
+    int32_t index_length = GetIndexLength(footer, 0);
     PAIMON_RETURN_NOT_OK(input_stream->Seek(
-        file_size - BlobDefs::kBlobFileHeaderLength - index_length, FS_SEEK_SET));
+        file_size - BlobDefs::kBlobFileFooterLength - index_length, FS_SEEK_SET));
     std::vector<char> index_bytes(index_length, '\0');
     PAIMON_ASSIGN_OR_RAISE(actual_size, input_stream->Read(index_bytes.data(), index_length));
     if (actual_size != index_length) {

src/paimon/format/blob/blob_format_writer.cpp

@@ -47,6 +47,7 @@ BlobFormatWriter::BlobFormatWriter(const std::shared_ptr<OutputStream>& out, con
       write_consumer_(std::move(write_consumer)) {
     metrics_ = std::make_shared<MetricsImpl>();
     tmp_buffer_ = Bytes::AllocateBytes(kTmpBufferSize, pool_.get());
+    magic_number_bytes_ = IntegerToLittleEndian<int32_t>(BlobDefs::kMagicNumber, pool_);
 }
 
 Result<std::unique_ptr<BlobFormatWriter>> BlobFormatWriter::Create(
@@ -161,9 +162,7 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
     PAIMON_ASSIGN_OR_RAISE(int64_t previous_pos, out_->GetPos());
 
     // write magic number
-    static PAIMON_UNIQUE_PTR<Bytes> kMagicNumberBytes =
-        IntegerToLittleEndian<int32_t>(BlobDefs::kMagicNumber, pool_);
-    PAIMON_RETURN_NOT_OK(WriteWithCrc32(kMagicNumberBytes->data(), kMagicNumberBytes->size()));
+    PAIMON_RETURN_NOT_OK(WriteWithCrc32(magic_number_bytes_->data(), magic_number_bytes_->size()));
 
     // write blob content
     // Dynamically check whether blob_data is a serialized BlobDescriptor (by magic header)
@@ -185,8 +184,9 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
     while (read_len > 0) {
         PAIMON_ASSIGN_OR_RAISE(int64_t actual_read_len, in->Read(tmp_buffer_->data(), read_len));
         if (actual_read_len != read_len) {
-            return Status::Invalid("actual read length {}, not match with expect length {}",
-                                   actual_read_len, read_len);
+            return Status::Invalid(
+                fmt::format("actual read length {}, not match with expect length {}",
+                            actual_read_len, read_len));
         }
         PAIMON_RETURN_NOT_OK(WriteWithCrc32(tmp_buffer_->data(), actual_read_len));
         total_read_length += actual_read_len;
@@ -214,8 +214,8 @@ Status BlobFormatWriter::WriteBlob(std::string_view blob_data) {
 Status BlobFormatWriter::WriteBytes(const char* data, int64_t length) {
     PAIMON_ASSIGN_OR_RAISE(int64_t actual, out_->Write(data, length));
     if (actual != length) {
-        return Status::Invalid("not suppose actual length {} not match with expect {}", actual,
-                               length);
+        return Status::Invalid(
+            fmt::format("unexpected actual length {} not match with expect {}", actual, length));
     }
     return Status::OK();
 }

src/paimon/format/blob/blob_format_writer.h

@@ -90,15 +90,15 @@ class BlobFormatWriter : public FormatWriter {
     static PAIMON_UNIQUE_PTR<Bytes> IntegerToLittleEndian(T value,
                                                           const std::shared_ptr<MemoryPool>& pool);
 
- public:
+ private:
     static constexpr uint32_t kTmpBufferSize = 1024 * 1024;
 
- private:
     uint32_t crc32_ = 0;
     std::vector<int64_t> bin_lengths_;
     std::shared_ptr<OutputStream> out_;
     std::string uri_;
     PAIMON_UNIQUE_PTR<Bytes> tmp_buffer_;
+    PAIMON_UNIQUE_PTR<Bytes> magic_number_bytes_;
     std::shared_ptr<arrow::DataType> data_type_;
     std::shared_ptr<FileSystem> fs_;
     std::shared_ptr<MemoryPool> pool_;

src/paimon/format/blob/blob_stats_extractor.cpp

@@ -54,7 +54,7 @@ BlobStatsExtractor::ExtractWithFileInfo(const std::shared_ptr<FileSystem>& file_
                                     /*batch_size=*/1024, /*blob_as_descriptor=*/true, pool));
     ColumnStatsVector result_stats;
     result_stats.push_back(
-        ColumnStats::CreateStringColumnStats(std::nullopt, std::nullopt, /*null_count=*/0));
+        ColumnStats::CreateStringColumnStats(std::nullopt, std::nullopt, std::nullopt));
     PAIMON_ASSIGN_OR_RAISE(uint64_t num_rows, blob_reader->GetNumberOfRows());
     return std::make_pair(result_stats, FileInfo(num_rows));
 }