feat: Add basic support for writing to an Iceberg table (facebookincubator#14723)

Summary:
As per review request from facebookincubator#10996, we should split facebookincubator#10996 to multiple smaller PRs.

This is the first PR of them. It adds support for basic insertion without partition transforms or data file statistics. The implementation supports both primitive data types and nested types, including struct, map, and array.

A series of follow-up PRs will extend this work to make Iceberg write fully compatible with the Iceberg specification.

1. Iceberg partition spec (new files).
2. Add new option for Timestamp::tmToStringView to meet Iceberg spec (completely standalone code).
3. Customise iceberg writer options.
4. Iceberg file name generator.
5. Iceberg partition ID generator (new files).
6. Identity partition transform.

Pull Request resolved: facebookincubator#14723

Reviewed By: xiaoxmeng

Differential Revision: D83667667

Pulled By: kgpai

fbshipit-source-id: a3df1aeac432d6dde6610b9a7057e170a706de9e

Author: PingLiuPing

velox/connectors/hive/HiveConnector.cpp

@@ -20,6 +20,7 @@
 #include "velox/connectors/hive/HiveDataSink.h"
 #include "velox/connectors/hive/HiveDataSource.h"
 #include "velox/connectors/hive/HivePartitionFunction.h"
+#include "velox/connectors/hive/iceberg/IcebergDataSink.h"
 
 #include <boost/lexical_cast.hpp>
 #include <memory>
@@ -73,6 +74,16 @@ std::unique_ptr<DataSink> HiveConnector::createDataSink(
     ConnectorInsertTableHandlePtr connectorInsertTableHandle,
     ConnectorQueryCtx* connectorQueryCtx,
     CommitStrategy commitStrategy) {
+  if (auto icebergInsertHandle =
+          std::dynamic_pointer_cast<const iceberg::IcebergInsertTableHandle>(
+              connectorInsertTableHandle)) {
+    return std::make_unique<iceberg::IcebergDataSink>(
+        inputType,
+        icebergInsertHandle,
+        connectorQueryCtx,
+        commitStrategy,
+        hiveConfig_);
+  }
   auto hiveInsertHandle =
       std::dynamic_pointer_cast<const HiveInsertTableHandle>(
           connectorInsertTableHandle);

velox/connectors/hive/HiveDataSink.cpp

@@ -682,7 +682,10 @@ bool HiveDataSink::finish() {
 std::vector<std::string> HiveDataSink::close() {
   setState(State::kClosed);
   closeInternal();
+  return commitMessage();
+}
 
+std::vector<std::string> HiveDataSink::commitMessage() const {
   std::vector<std::string> partitionUpdates;
   partitionUpdates.reserve(writerInfo_.size());
   for (int i = 0; i < writerInfo_.size(); ++i) {

velox/connectors/hive/HiveDataSink.h

@@ -337,9 +337,11 @@ class HiveInsertTableHandle : public ConnectorInsertTableHandle {
 
   std::string toString() const override;
 
- private:
+ protected:
   const std::vector<std::shared_ptr<const HiveColumnHandle>> inputColumns_;
   const std::shared_ptr<const LocationHandle> locationHandle_;
+
+ private:
   const dwio::common::FileFormat storageFormat_;
   const std::shared_ptr<const HiveBucketProperty> bucketProperty_;
   const std::optional<common::CompressionKind> compressionKind_;
@@ -544,11 +546,19 @@ class HiveDataSink : public DataSink {
 
   bool canReclaim() const;
 
- private:
+ protected:
   // Validates the state transition from 'oldState' to 'newState'.
   void checkStateTransition(State oldState, State newState);
   void setState(State newState);
 
+  // Generates commit messages for all writers containing metadata about written
+  // files. Creates a JSON object for each writer with partition name,
+  // file paths, file names, data sizes, and row counts. This metadata is used
+  // by the coordinator to commit the transaction and update the metastore.
+  //
+  // @return Vector of JSON strings, one per writer.
+  virtual std::vector<std::string> commitMessage() const;
+
   class WriterReclaimer : public exec::MemoryReclaimer {
    public:
     static std::unique_ptr<memory::MemoryReclaimer> create(

velox/connectors/hive/iceberg/CMakeLists.txt

@@ -14,6 +14,7 @@
 
 velox_add_library(
   velox_hive_iceberg_splitreader
+  IcebergDataSink.cpp
   IcebergSplitReader.cpp
   IcebergSplit.cpp
   PositionalDeleteFileReader.cpp

velox/connectors/hive/iceberg/IcebergDataSink.cpp

@@ -0,0 +1,88 @@
+/*
+ * 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/IcebergDataSink.h"
+#include "velox/common/base/Fs.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+IcebergInsertTableHandle::IcebergInsertTableHandle(
+    std::vector<HiveColumnHandlePtr> inputColumns,
+    LocationHandlePtr locationHandle,
+    dwio::common::FileFormat tableStorageFormat,
+    std::optional<common::CompressionKind> compressionKind,
+    const std::unordered_map<std::string, std::string>& serdeParameters)
+    : HiveInsertTableHandle(
+          std::move(inputColumns),
+          std::move(locationHandle),
+          tableStorageFormat,
+          nullptr,
+          compressionKind,
+          serdeParameters,
+          nullptr,
+          false,
+          std::make_shared<const HiveInsertFileNameGenerator>()) {
+  VELOX_USER_CHECK(
+      !inputColumns_.empty(),
+      "Input columns cannot be empty for Iceberg tables.");
+  VELOX_USER_CHECK_NOT_NULL(
+      locationHandle_, "Location handle is required for Iceberg tables.");
+}
+
+IcebergDataSink::IcebergDataSink(
+    RowTypePtr inputType,
+    IcebergInsertTableHandlePtr insertTableHandle,
+    const ConnectorQueryCtx* connectorQueryCtx,
+    CommitStrategy commitStrategy,
+    const std::shared_ptr<const HiveConfig>& hiveConfig)
+    : HiveDataSink(
+          std::move(inputType),
+          insertTableHandle,
+          connectorQueryCtx,
+          commitStrategy,
+          hiveConfig,
+          0,
+          nullptr) {}
+
+std::vector<std::string> IcebergDataSink::commitMessage() const {
+  std::vector<std::string> commitTasks;
+  commitTasks.reserve(writerInfo_.size());
+
+  for (auto i = 0; i < writerInfo_.size(); ++i) {
+    const auto& info = writerInfo_.at(i);
+    VELOX_CHECK_NOT_NULL(info);
+    // Following metadata (json format) is consumed by Presto CommitTaskData.
+    // It contains the minimal subset of metadata.
+    // TODO: Complete metrics is missing now and this could lead to suboptimal
+    // query plan, will collect full iceberg metrics in following PR.
+    // clang-format off
+    folly::dynamic commitData = folly::dynamic::object(
+    "path", (fs::path(info->writerParameters.writeDirectory()) /
+                    info->writerParameters.writeFileName()).string())
+      ("fileSizeInBytes", ioStats_.at(i)->rawBytesWritten())
+      ("metrics",
+        folly::dynamic::object("recordCount", info->numWrittenRows))
+      ("partitionSpecJson", 0)
+      ("fileFormat", "PARQUET")
+      ("content", "DATA");
+    // clang-format on
+    auto commitDataJson = folly::toJson(commitData);
+    commitTasks.push_back(commitDataJson);
+  }
+  return commitTasks;
+}
+
+} // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/IcebergDataSink.h

@@ -0,0 +1,81 @@
+/*
+ * 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 "velox/connectors/hive/HiveDataSink.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+/// Represents a request for Iceberg write.
+class IcebergInsertTableHandle final : public HiveInsertTableHandle {
+ public:
+  /// @param inputColumns Columns from the table schema to write.
+  /// The input RowVector must have the same number of columns and matching
+  /// types in the same order.
+  /// Column names in the RowVector may differ from those in inputColumns,
+  /// only position and type must align. All columns present in the input
+  /// data must be included, mismatches can lead to write failure.
+  /// @param locationHandle Contains the target location information including:
+  /// - Base directory path where data files will be written.
+  /// - File naming scheme and temporary directory paths.
+  /// @param compressionKind Optional compression to apply to data files.
+  /// @param serdeParameters Additional serialization/deserialization parameters
+  /// for the file format.
+  IcebergInsertTableHandle(
+      std::vector<HiveColumnHandlePtr> inputColumns,
+      LocationHandlePtr locationHandle,
+      dwio::common::FileFormat tableStorageFormat,
+      std::optional<common::CompressionKind> compressionKind = {},
+      const std::unordered_map<std::string, std::string>& serdeParameters = {});
+};
+
+using IcebergInsertTableHandlePtr =
+    std::shared_ptr<const IcebergInsertTableHandle>;
+
+class IcebergDataSink : public HiveDataSink {
+ public:
+  IcebergDataSink(
+      RowTypePtr inputType,
+      IcebergInsertTableHandlePtr insertTableHandle,
+      const ConnectorQueryCtx* connectorQueryCtx,
+      CommitStrategy commitStrategy,
+      const std::shared_ptr<const HiveConfig>& hiveConfig);
+
+  /// Generates Iceberg-specific commit messages for all writers containing
+  /// metadata about written files. Creates a JSON object for each writer
+  /// in the format expected by Presto and Spark for Iceberg tables.
+  ///
+  /// Each commit message contains:
+  /// - path: full file path where data was written.
+  /// - fileSizeInBytes: raw bytes written to disk.
+  /// - metrics: object with recordCount (number of rows written).
+  /// - partitionSpecJson: partition specification.
+  /// - fileFormat: storage format (e.g., "PARQUET").
+  /// - content: file content type ("DATA" for data files).
+  ///
+  /// See
+  /// https://github.com/prestodb/presto/blob/master/presto-iceberg/src/main/java/com/facebook/presto/iceberg/CommitTaskData.java
+  ///
+  /// Note: Complete Iceberg metrics are not yet implemented, which results in
+  /// incomplete manifest files that may lead to suboptimal query planning.
+  ///
+  /// @return Vector of JSON strings, one per writer, formatted according to
+  /// Presto and Spark Iceberg commit protocol.
+  std::vector<std::string> commitMessage() const override;
+};
+
+} // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/tests/CMakeLists.txt

@@ -56,9 +56,29 @@ if(NOT VELOX_DISABLE_GOOGLETEST)
     GTest::gtest
     GTest::gtest_main
   )
+
+  add_executable(velox_hive_iceberg_insert_test IcebergInsertTest.cpp IcebergTestBase.cpp Main.cpp)
+
+  add_test(velox_hive_iceberg_insert_test velox_hive_iceberg_insert_test)
+
+  target_link_libraries(
+    velox_hive_iceberg_insert_test
+    velox_exec_test_lib
+    velox_hive_connector
+    velox_hive_iceberg_splitreader
+    velox_vector_fuzzer
+    GTest::gtest
+  )
+
   if(VELOX_ENABLE_PARQUET)
     target_link_libraries(velox_hive_iceberg_test velox_dwio_parquet_reader)
 
     file(COPY ${CMAKE_CURRENT_SOURCE_DIR}/examples DESTINATION ${CMAKE_CURRENT_BINARY_DIR})
+
+    target_link_libraries(
+      velox_hive_iceberg_insert_test
+      velox_dwio_parquet_reader
+      velox_dwio_parquet_writer
+    )
   endif()
 endif()

velox/connectors/hive/iceberg/tests/IcebergInsertTest.cpp

@@ -0,0 +1,81 @@
+/*
+ * 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/tests/IcebergTestBase.h"
+#include "velox/exec/tests/utils/PlanBuilder.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+namespace {
+
+class IcebergInsertTest : public test::IcebergTestBase {
+ protected:
+  void test(const RowTypePtr& rowType, double nullRatio = 0.0) {
+    const auto outputDirectory = exec::test::TempDirectoryPath::create();
+    const auto dataPath = fmt::format("{}", outputDirectory->getPath());
+    constexpr int32_t numBatches = 10;
+    constexpr int32_t vectorSize = 5'000;
+    const auto vectors =
+        createTestData(rowType, numBatches, vectorSize, nullRatio);
+    auto dataSink =
+        createIcebergDataSink(rowType, outputDirectory->getPath(), {});
+
+    for (const auto& vector : vectors) {
+      dataSink->appendData(vector);
+    }
+
+    ASSERT_TRUE(dataSink->finish());
+    const auto commitTasks = dataSink->close();
+    createDuckDbTable(vectors);
+    auto splits = createSplitsForDirectory(dataPath);
+    ASSERT_EQ(splits.size(), commitTasks.size());
+    auto plan = exec::test::PlanBuilder().tableScan(rowType).planNode();
+    assertQuery(plan, splits, "SELECT * FROM tmp");
+  }
+};
+
+TEST_F(IcebergInsertTest, basic) {
+  auto rowType =
+      ROW({"c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9", "c10", "c11"},
+          {BIGINT(),
+           INTEGER(),
+           SMALLINT(),
+           BOOLEAN(),
+           REAL(),
+           DECIMAL(18, 5),
+           VARCHAR(),
+           VARBINARY(),
+           DATE(),
+           TIMESTAMP(),
+           ROW({"id", "name"}, {INTEGER(), VARCHAR()})});
+  test(rowType, 0.2);
+}
+
+TEST_F(IcebergInsertTest, mapAndArray) {
+  auto rowType =
+      ROW({"c1", "c2"}, {MAP(INTEGER(), VARCHAR()), ARRAY(VARCHAR())});
+  test(rowType);
+}
+
+#ifdef VELOX_ENABLE_PARQUET
+TEST_F(IcebergInsertTest, bigDecimal) {
+  auto rowType = ROW({"c1"}, {DECIMAL(38, 5)});
+  fileFormat_ = dwio::common::FileFormat::PARQUET;
+  test(rowType);
+}
+#endif
+
+} // namespace
+} // namespace facebook::velox::connector::hive::iceberg