add file metric in writer

Author: xiao.dong

src/iceberg/avro/avro_writer.cc

@@ -85,6 +85,7 @@ class AvroWriter::Impl {
       ICEBERG_RETURN_UNEXPECTED(InitWriteContext());
     }
     // TODO(xiao.dong) convert data and write to avro
+    // total_bytes_+= written_bytes;
     return {};
   }
 
@@ -97,10 +98,15 @@ class AvroWriter::Impl {
     return {};
   }
 
+  bool Closed() const { return writer_ == nullptr; }
+
+  int64_t length() { return total_bytes_; }
+
  private:
   Status InitWriteContext() { return {}; }
 
  private:
+  int64_t total_bytes_ = 0;
   // The schema to write.
   std::shared_ptr<::iceberg::Schema> write_schema_;
   // The avro schema to write.
@@ -120,7 +126,29 @@ Status AvroWriter::Open(const WriterOptions& options) {
   return impl_->Open(options);
 }
 
-Status AvroWriter::Close() { return impl_->Close(); }
+Status AvroWriter::Close() {
+  if (!impl_->Closed()) {
+    return impl_->Close();
+  }
+  return {};
+}
+
+std::shared_ptr<Metrics> AvroWriter::metrics() {
+  if (impl_->Closed()) {
+    // TODO(xiao.dong) implement metrics
+    return std::make_shared<Metrics>();
+  }
+  return nullptr;
+}
+
+int64_t AvroWriter::length() {
+  if (impl_->Closed()) {
+    return impl_->length();
+  }
+  return 0;
+}
+
+std::vector<int64_t> AvroWriter::splitOffsets() { return {}; }
 
 void AvroWriter::Register() {
   static WriterFactoryRegistry avro_writer_register(

src/iceberg/avro/avro_writer.h

@@ -37,6 +37,12 @@ class ICEBERG_BUNDLE_EXPORT AvroWriter : public Writer {
 
   Status Write(ArrowArray data) final;
 
+  std::shared_ptr<Metrics> metrics() final;
+
+  int64_t length() final;
+
+  std::vector<int64_t> splitOffsets() final;
+
   /// \brief Register this Avro writer implementation.
   static void Register();
 

src/iceberg/file_writer.h

@@ -28,6 +28,7 @@
 
 #include "iceberg/arrow_c_data.h"
 #include "iceberg/file_format.h"
+#include "iceberg/metrics.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/type_fwd.h"
@@ -66,6 +67,19 @@ class ICEBERG_EXPORT Writer {
   ///
   /// \return Status of write results.
   virtual Status Write(ArrowArray data) = 0;
+
+  /// \brief Get the file statistics.
+  virtual std::shared_ptr<Metrics> metrics() = 0;
+
+  /// \brief Get the file length.
+  virtual int64_t length() = 0;
+
+  /// \brief Get the file length.
+  /// Returns a list of recommended split locations, if applicable, null otherwise.
+  /// When available, this information is used for planning scan tasks whose boundaries
+  /// are determined by these offsets. The returned list must be sorted in ascending order
+  /// Only valid after the file is closed.
+  virtual std::vector<int64_t> splitOffsets() = 0;
 };
 
 /// \brief Factory function to create a writer of a specific file format.

src/iceberg/manifest_writer.h

@@ -35,9 +35,16 @@ namespace iceberg {
 class ICEBERG_EXPORT ManifestWriter {
  public:
   virtual ~ManifestWriter() = default;
+
+  /// \brief Write manifest entries to file
+  /// \param entries List of manifest entries to write.
+  /// \return Status::OK() if all entries were written successfully
   virtual Status WriteManifestEntries(
       const std::vector<ManifestEntry>& entries) const = 0;
 
+  /// \brief Close writer and flush to storage.
+  virtual void Close() = 0;
+
   /// \brief Creates a writer for a manifest file.
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
@@ -51,8 +58,15 @@ class ICEBERG_EXPORT ManifestWriter {
 class ICEBERG_EXPORT ManifestListWriter {
  public:
   virtual ~ManifestListWriter() = default;
+
+  /// \brief Write manifest file list to mainifest list file.
+  /// \param files List of manifest files to write.
+  /// \return Status::OK() if all files were written successfully
   virtual Status WriteManifestFiles(const std::vector<ManifestFile>& files) const = 0;
 
+  /// \brief Close writer and flush to storage.
+  virtual void Close() = 0;
+
   /// \brief Creates a writer for the manifest list.
   /// \param manifest_list_location Path to the manifest list file.
   /// \param file_io File IO implementation to use.

src/iceberg/manifest_writer_internal.h

@@ -35,6 +35,8 @@ class ManifestWriterImpl : public ManifestWriter {
 
   Status WriteManifestEntries(const std::vector<ManifestEntry>& entries) const override;
 
+  void Close() override {}
+
  private:
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Writer> writer_;
@@ -49,6 +51,8 @@ class ManifestListWriterImpl : public ManifestListWriter {
 
   Status WriteManifestFiles(const std::vector<ManifestFile>& files) const override;
 
+  void Close() override {}
+
  private:
   std::shared_ptr<Schema> schema_;
   std::unique_ptr<Writer> writer_;

src/iceberg/metrics.h

@@ -0,0 +1,60 @@
+/*
+ * 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/metrics.h
+
+#include <unordered_map>
+
+#include "iceberg/expression/literal.h"
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Iceberg file format metrics
+class ICEBERG_EXPORT Metrics {
+ public:
+  Metrics() = default;
+
+  Metrics(int64_t row_count, std::unordered_map<int64_t, int64_t> column_sizes = {},
+          std::unordered_map<int64_t, int64_t> value_counts = {},
+          std::unordered_map<int64_t, int64_t> null_value_counts = {},
+          std::unordered_map<int64_t, int64_t> nan_value_counts = {},
+          std::unordered_map<int64_t, Literal> lower_bounds = {},
+          std::unordered_map<int64_t, Literal> upper_bounds = {})
+      : row_count_(row_count),
+        column_sizes_(std::move(column_sizes)),
+        value_counts_(std::move(value_counts)),
+        null_value_counts_(std::move(null_value_counts)),
+        nan_value_counts_(std::move(nan_value_counts)),
+        lower_bounds_(std::move(lower_bounds)),
+        upper_bounds_(std::move(upper_bounds)) {}
+
+ private:
+  int64_t row_count_ = 0;
+  std::unordered_map<int64_t, int64_t> column_sizes_;
+  std::unordered_map<int64_t, int64_t> value_counts_;
+  std::unordered_map<int64_t, int64_t> null_value_counts_;
+  std::unordered_map<int64_t, int64_t> nan_value_counts_;
+  std::unordered_map<int64_t, Literal> lower_bounds_;
+  std::unordered_map<int64_t, Literal> upper_bounds_;
+};
+
+}  // namespace iceberg