fix review 2

Author: zjw1111

src/paimon/format/avro/avro_file_format_factory.cpp

@@ -27,11 +27,10 @@ const char AvroFileFormatFactory::IDENTIFIER[] = "avro";
 
 Result<std::unique_ptr<FileFormat>> AvroFileFormatFactory::Create(
     const std::map<std::string, std::string>& options) const {
-    RegisterLogicalTypes();
     return std::make_unique<AvroFileFormat>(options);
 }
 
-void AvroFileFormatFactory::RegisterLogicalTypes() {
+static __attribute__((constructor)) void AvroFileFormatFactoryRegisterLogicalTypes() {
     ::avro::CustomLogicalTypeRegistry::instance().registerType(
         "map", [](const std::string&) { return std::make_shared<MapLogicalType>(); });
 }

src/paimon/format/avro/avro_file_format_factory.h

@@ -41,9 +41,6 @@ class AvroFileFormatFactory : public FileFormatFactory {
 
     Result<std::unique_ptr<FileFormat>> Create(
         const std::map<std::string, std::string>& options) const override;
-
- private:
-    static void RegisterLogicalTypes();
 };
 
 }  // namespace paimon::avro

src/paimon/format/avro/avro_format_writer.cpp

@@ -39,34 +39,33 @@
 namespace arrow {
 class Array;
 }  // namespace arrow
-namespace avro {
-class OutputStream;
-}  // namespace avro
 struct ArrowArray;
 
 namespace paimon::avro {
 
 AvroFormatWriter::AvroFormatWriter(
     const std::shared_ptr<::avro::DataFileWriter<::avro::GenericDatum>>& file_writer,
     const ::avro::ValidSchema& avro_schema, const std::shared_ptr<arrow::DataType>& data_type,
-    std::unique_ptr<AvroAdaptor> adaptor)
+    std::unique_ptr<AvroAdaptor> adaptor, AvroOutputStreamImpl* avro_output_stream)
     : writer_(file_writer),
       avro_schema_(avro_schema),
       data_type_(data_type),
-      adaptor_(std::move(adaptor)) {}
+      adaptor_(std::move(adaptor)),
+      avro_output_stream_(avro_output_stream) {}
 
 Result<std::unique_ptr<AvroFormatWriter>> AvroFormatWriter::Create(
-    std::unique_ptr<::avro::OutputStream> out, const std::shared_ptr<arrow::Schema>& schema,
+    std::unique_ptr<AvroOutputStreamImpl> out, const std::shared_ptr<arrow::Schema>& schema,
     const ::avro::Codec codec) {
     try {
         PAIMON_ASSIGN_OR_RAISE(::avro::ValidSchema avro_schema,
                                AvroSchemaConverter::ArrowSchemaToAvroSchema(schema));
+        AvroOutputStreamImpl* avro_output_stream = out.get();
         auto writer = std::make_shared<::avro::DataFileWriter<::avro::GenericDatum>>(
             std::move(out), avro_schema, DEFAULT_SYNC_INTERVAL, codec);
         auto data_type = arrow::struct_(schema->fields());
         auto adaptor = std::make_unique<AvroAdaptor>(data_type);
-        return std::unique_ptr<AvroFormatWriter>(
-            new AvroFormatWriter(writer, avro_schema, data_type, std::move(adaptor)));
+        return std::unique_ptr<AvroFormatWriter>(new AvroFormatWriter(
+            writer, avro_schema, data_type, std::move(adaptor), avro_output_stream));
     } catch (const ::avro::Exception& e) {
         return Status::Invalid(fmt::format("avro format writer create failed. {}", e.what()));
     } catch (const std::exception& e) {
@@ -92,6 +91,7 @@ Status AvroFormatWriter::Flush() {
 
 Status AvroFormatWriter::Finish() {
     try {
+        avro_output_stream_->FlushBuffer();  // we need flush buffer before close writer
         writer_->close();
     } catch (const ::avro::Exception& e) {
         return Status::Invalid(fmt::format("avro writer close failed. {}", e.what()));

src/paimon/format/avro/avro_format_writer.h

@@ -22,9 +22,9 @@
 
 #include "arrow/api.h"
 #include "avro/DataFile.hh"
-#include "avro/Stream.hh"
 #include "avro/ValidSchema.hh"
 #include "paimon/format/avro/avro_adaptor.h"
+#include "paimon/format/avro/avro_output_stream_impl.h"
 #include "paimon/format/format_writer.h"
 #include "paimon/metrics.h"
 #include "paimon/result.h"
@@ -36,7 +36,6 @@ class Schema;
 }  // namespace arrow
 namespace avro {
 class GenericDatum;
-class OutputStream;
 }  // namespace avro
 namespace paimon {
 class Metrics;
@@ -49,7 +48,7 @@ namespace paimon::avro {
 class AvroFormatWriter : public FormatWriter {
  public:
     static Result<std::unique_ptr<AvroFormatWriter>> Create(
-        std::unique_ptr<::avro::OutputStream> out, const std::shared_ptr<arrow::Schema>& schema,
+        std::unique_ptr<AvroOutputStreamImpl> out, const std::shared_ptr<arrow::Schema>& schema,
         const ::avro::Codec codec);
 
     Status AddBatch(ArrowArray* batch) override;
@@ -70,13 +69,14 @@ class AvroFormatWriter : public FormatWriter {
     AvroFormatWriter(
         const std::shared_ptr<::avro::DataFileWriter<::avro::GenericDatum>>& file_writer,
         const ::avro::ValidSchema& avro_schema, const std::shared_ptr<arrow::DataType>& data_type,
-        std::unique_ptr<AvroAdaptor> adaptor);
+        std::unique_ptr<AvroAdaptor> adaptor, AvroOutputStreamImpl* avro_output_stream);
 
     std::shared_ptr<::avro::DataFileWriter<::avro::GenericDatum>> writer_;
     ::avro::ValidSchema avro_schema_;
     std::shared_ptr<arrow::DataType> data_type_;
     std::shared_ptr<Metrics> metrics_;
     std::unique_ptr<AvroAdaptor> adaptor_;
+    AvroOutputStreamImpl* avro_output_stream_;
 };
 
 }  // namespace paimon::avro

src/paimon/format/avro/avro_output_stream_impl.cpp

@@ -39,7 +39,6 @@ AvroOutputStreamImpl::AvroOutputStreamImpl(const std::shared_ptr<paimon::OutputS
       byte_count_(0) {}
 
 AvroOutputStreamImpl::~AvroOutputStreamImpl() {
-    FlushBuffer();
     pool_->Free(buffer_, buffer_size_);
 }
 

src/paimon/format/avro/avro_output_stream_impl.h

@@ -44,9 +44,9 @@ class AvroOutputStreamImpl : public ::avro::OutputStream {
         return byte_count_;
     }
 
- private:
     void FlushBuffer();
 
+ private:
     std::shared_ptr<MemoryPool> pool_;
     const size_t buffer_size_;
     uint8_t* const buffer_;

src/paimon/format/avro/avro_schema_converter.cpp

@@ -178,16 +178,16 @@ Result<std::shared_ptr<arrow::DataType>> AvroSchemaConverter::GetArrowType(
             if (fields.size() != 2) {
                 return Status::TypeError("invalid avro logical map struct fields size");
             }
-            auto key_field = fields[0];
-            key_field = key_field->WithNullable(false);
+            auto key_field = fields[0]->WithNullable(false);
             auto value_field = fields[1];
             if (key_field->name() != "key" || value_field->name() != "value") {
                 return Status::TypeError("invalid avro logical map struct field names");
             }
             return std::make_shared<arrow::MapType>(std::move(key_field), std::move(value_field));
         }
         default:
-            return Status::Invalid("not support logical type: ", AvroUtils::ToString(logical_type));
+            return Status::Invalid("invalid avro logical type: ",
+                                   AvroUtils::ToString(logical_type));
     }
 
     size_t subtype_count = avro_node->leaves();

src/paimon/format/orc/orc_adapter.cpp

@@ -856,7 +856,7 @@ Result<std::shared_ptr<arrow::ArrayBuilder>> MakeArrowBuilder(
     arrow::MemoryPool* pool) {
     if (column_vector_batch->numElements == 0) {
         PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::ArrayBuilder> builder,
-                                          arrow::MakeBuilder(type));
+                                          arrow::MakeBuilder(type, pool));
         return builder;
     }
     arrow::Type::type kind = type->id();