feat: Collect Iceberg stats

velox/connectors/hive/HiveDataSink.h

@@ -717,13 +717,12 @@ class HiveDataSink : public DataSink {
   // Sets up compression, schema, and other writer configuration based on the
   // insert table handle and connector settings.
   // The no-argument overload uses the last writer's info (for appendWriter).
-  virtual std::shared_ptr<dwio::common::WriterOptions> createWriterOptions()
-      const;
+  std::shared_ptr<dwio::common::WriterOptions> createWriterOptions() const;
 
   // Creates WriterOptions for a specific writer index. Use this overload
   // during writer rotation to ensure the correct writer's memory pool and
   // nonReclaimableSection are used.
-  std::shared_ptr<dwio::common::WriterOptions> createWriterOptions(
+  virtual std::shared_ptr<dwio::common::WriterOptions> createWriterOptions(
       size_t writerIndex) const;
 
   // Returns the Hive partition directory name for the given partition ID.
@@ -771,14 +770,14 @@ class HiveDataSink : public DataSink {
   /// Rotates the writer at the given index to a new file. This is called when
   /// the current file exceeds maxTargetFileBytes_. The old writer is closed
   /// and a new writer is created for the same partition/bucket.
-  void rotateWriter(size_t index);
+  virtual void rotateWriter(size_t index);
 
   /// Finalizes the current file for the writer at the given index.
   /// Captures file stats and adds the file info to writtenFiles.
   /// Called by rotateWriter() and closeInternal().
   void finalizeWriterFile(size_t index);
 
-  void closeInternal();
+  virtual void closeInternal();
 
   // IMPORTANT NOTE: these are passed to writers as raw pointers. HiveDataSink
   // owns the lifetime of these objects, and therefore must destroy them last.

velox/connectors/hive/iceberg/CMakeLists.txt

@@ -12,11 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-velox_add_library(
-  velox_hive_iceberg_splitreader
+set(
+  ICEBERG_SOURCES
   IcebergConfig.cpp
   IcebergColumnHandle.cpp
   IcebergConnector.cpp
+  IcebergDataFileStatistics.cpp
   IcebergDataSink.cpp
   IcebergDataSource.cpp
   IcebergPartitionName.cpp
@@ -28,17 +29,20 @@ velox_add_library(
   TransformExprBuilder.cpp
 )
 
+if(VELOX_ENABLE_PARQUET)
+  list(APPEND ICEBERG_SOURCES IcebergParquetStatsCollector.cpp)
+endif()
+
+velox_add_library(velox_hive_iceberg_splitreader ${ICEBERG_SOURCES})
+
 velox_link_libraries(
   velox_hive_iceberg_splitreader
   velox_connector
+  velox_dwio_parquet_field_id
   velox_functions_iceberg
   Folly::folly
 )
 
-if(VELOX_ENABLE_PARQUET)
-  velox_link_libraries(velox_hive_iceberg_splitreader velox_dwio_parquet_field_id)
-endif()
-
 if(${VELOX_BUILD_TESTING})
   add_subdirectory(tests)
 endif()

velox/connectors/hive/iceberg/IcebergDataFileStatistics.cpp

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+
+#include "velox/connectors/hive/iceberg/IcebergDataFileStatistics.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+folly::dynamic IcebergDataFileStatistics::toJson() const {
+  folly::dynamic json = folly::dynamic::object;
+  json["recordCount"] = numRecords;
+
+  folly::dynamic columnSizes = folly::dynamic::object;
+  folly::dynamic valueCounts = folly::dynamic::object;
+  folly::dynamic nullValueCounts = folly::dynamic::object;
+  folly::dynamic nanValueCounts = folly::dynamic::object;
+  folly::dynamic lowerBounds = folly::dynamic::object;
+  folly::dynamic upperBounds = folly::dynamic::object;
+
+  for (const auto& [fieldId, stats] : columnStats) {
+    auto fieldIdStr = folly::to<std::string>(fieldId);
+    columnSizes[fieldIdStr] = stats.columnSize;
+    valueCounts[fieldIdStr] = stats.valueCount;
+    nullValueCounts[fieldIdStr] = stats.nullValueCount;
+    if (stats.nanValueCount.has_value()) {
+      nanValueCounts[fieldIdStr] = stats.nanValueCount.value();
+    }
+    if (stats.lowerBound.has_value()) {
+      lowerBounds[fieldIdStr] = stats.lowerBound.value();
+    }
+    if (stats.upperBound.has_value()) {
+      upperBounds[fieldIdStr] = stats.upperBound.value();
+    }
+  }
+
+  json["columnSizes"] = std::move(columnSizes);
+  json["valueCounts"] = std::move(valueCounts);
+  json["nullValueCounts"] = std::move(nullValueCounts);
+  json["nanValueCounts"] = std::move(nanValueCounts);
+  json["lowerBounds"] = std::move(lowerBounds);
+  json["upperBounds"] = std::move(upperBounds);
+
+  return json;
+}
+
+} // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/IcebergDataFileStatistics.h

@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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
+
+#include <cstdint>
+#include <optional>
+#include <string>
+
+#include <folly/container/F14Map.h>
+#include <folly/json/dynamic.h>
+
+namespace facebook::velox::connector::hive::iceberg {
+
+/// Statistics for an Iceberg data file, corresponding to the `data_file`
+/// structure defined in the Iceberg specification:
+/// https://iceberg.apache.org/spec/#data-file-fields.
+///
+/// All column-level statistics maps are keyed by Iceberg field IDs (`int32_t`),
+/// which uniquely identify columns in the Iceberg schema independent of column
+/// names or physical column positions.
+struct IcebergDataFileStatistics {
+  struct ColumnStats {
+    int64_t columnSize{0};
+
+    /// Total number of values for this field ID in the file, including null and
+    /// NaN values.
+    ///
+    /// For primitive (flat) columns, this is equal to the number of rows in the
+    /// file: numRows = valueCount = (nonNullValues + numNulls + numNaNs).
+    ///
+    /// For nested columns (e.g. elements inside an array), this represents the
+    /// total occurrences of the field across all rows, which is not necessarily
+    /// related to the top-level record count.
+    int64_t valueCount{0};
+    int64_t nullValueCount{0};
+    std::optional<int64_t> nanValueCount;
+    /// Base64 encoded lower bound.
+    std::optional<std::string> lowerBound;
+    /// Base64 encoded upper bound.
+    std::optional<std::string> upperBound;
+  };
+
+  int64_t numRecords{0};
+  folly::F14FastMap<int32_t, ColumnStats> columnStats;
+
+  /// Returns a IcebergDataFileStatistics with all values set to zero/empty.
+  /// Useful for empty data files that have no actual data.
+  static IcebergDataFileStatistics empty() {
+    return IcebergDataFileStatistics{.numRecords = 0, .columnStats = {}};
+  }
+
+  folly::dynamic toJson() const;
+};
+
+using IcebergDataFileStatisticsPtr = std::shared_ptr<IcebergDataFileStatistics>;
+
+} // namespace facebook::velox::connector::hive::iceberg