Test

Author: infvg

backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala

@@ -327,8 +327,7 @@ class VeloxIcebergSuite extends IcebergSuite {
       val lastExecId = statusStore.executionsList().last.executionId
       val executionMetrics = statusStore.executionMetrics(lastExecId)
 
-      // TODO: fix https://github.com/apache/gluten/issues/11510
-      assert(executionMetrics(metrics("numWrittenFiles").id).toLong == 0)
+      assert(executionMetrics(metrics("numWrittenFiles").id).toLong == 1)
     }
   }
 
@@ -465,81 +464,4 @@ class VeloxIcebergSuite extends IcebergSuite {
       )
     }
   }
-  test("iceberg show stats has non-empty min max for store sales table") {
-    withTable("store_sales_10_rows") {
-      spark.sql("""
-                  |CREATE TABLE store_sales_10_rows (
-                  |  ss_sold_date_sk INT,
-                  |  ss_sold_time_sk INT,
-                  |  ss_item_sk INT,
-                  |  ss_customer_sk INT,
-                  |  ss_cdemo_sk INT,
-                  |  ss_hdemo_sk INT,
-                  |  ss_addr_sk INT,
-                  |  ss_store_sk INT,
-                  |  ss_promo_sk INT,
-                  |  ss_ticket_number BIGINT,
-                  |  ss_quantity INT,
-                  |  ss_wholesale_cost DECIMAL(7,2),
-                  |  ss_list_price DECIMAL(7,2),
-                  |  ss_sales_price DECIMAL(7,2),
-                  |  ss_ext_discount_amt DECIMAL(7,2),
-                  |  ss_ext_sales_price DECIMAL(7,2),
-                  |  ss_ext_wholesale_cost DECIMAL(7,2),
-                  |  ss_ext_list_price DECIMAL(7,2),
-                  |  ss_ext_tax DECIMAL(7,2),
-                  |  ss_coupon_amt DECIMAL(7,2),
-                  |  ss_net_paid DECIMAL(7,2),
-                  |  ss_net_paid_inc_tax DECIMAL(7,2),
-                  |  ss_net_profit DECIMAL(7,2)
-                  |) USING iceberg
-                  |""".stripMargin)
-
-      spark.sql(
-        """
-          |INSERT INTO store_sales_10_rows VALUES
-          |(2450899, null, 174781, null, null, 5105, 712262, null, null, 875206344, null, 75.64, 105.13, null, 0.00, null, 3328.16, 4625.72, null, 0.00, null, null, null),
-          |(2450899, 45381, 240260, 63498438, 1296795, 5105, 712262, 542, 1925, 875206344, 13, 5.12, 7.27, 2.18, 0.00, 28.34, 66.56, 94.51, 1.98, 0.00, 28.34, 30.32, -38.22),
-          |(2450899, 45381, 360506, 63498438, 1296795, 5105, 712262, 542, 332, 875206344, 69, 36.45, 70.34, 16.17, 0.00, 1115.73, 2515.05, 4853.46, 22.31, 0.00, 1115.73, 1138.04, -1399.32),
-          |(2450899, 45381, 197360, 63498438, 1296795, 5105, 712262, 542, 1486, 875206344, 50, 92.87, 167.16, 58.50, 0.00, 2925.00, 4643.50, 8358.00, 204.75, 0.00, 2925.00, 3129.75, -1718.50),
-          |(2450899, 45381, 58255, 63498438, 1296795, 5105, 712262, 542, 359, 875206344, 100, 85.99, 105.76, 47.59, 523.49, 4759.00, 8599.00, 10576.00, 296.48, 523.49, 4235.51, 4531.99, -4363.49),
-          |(2450899, 45381, 219500, 63498438, 1296795, 5105, 712262, 542, 8, 875206344, 77, 80.61, 121.72, 26.77, 2020.06, 2061.29, 6206.97, 9372.44, 1.64, 2020.06, 41.23, 42.87, -6165.74),
-          |(2450899, 45381, 60157, 63498438, 1296795, 5105, 712262, 542, 484, 875206344, 9, 44.58, 62.85, 50.28, 0.00, 452.52, 401.22, 565.65, 27.15, 0.00, 452.52, 479.67, 51.30),
-          |(2450899, 45381, 132362, 63498438, 1296795, 5105, 712262, 542, 1575, 875206344, 86, 30.79, 31.71, 25.68, 0.00, 2208.48, 2647.94, 2727.06, 22.08, 0.00, 2208.48, 2230.56, -439.46),
-          |(2450899, 45381, 41590, 63498438, 1296795, 5105, 712262, 542, 441, 875206344, 40, 79.99, 137.58, 12.38, 0.00, 495.20, 3199.60, 5503.20, 9.90, 0.00, 495.20, 505.10, -2704.40)
-          |""".stripMargin)
-
-      spark.sql("ANALYZE store_sales_10_rows")
-
-      withSQLConf("spark.sql.statistics.ignoreStatsCalculatorFailures" -> "false") {
-        val stats = spark.sql("SHOW STATS FOR store_sales_10_rows")
-
-        val statsByColumn =
-          stats.collect().map(row => row.getAs[String]("column_name") -> row).toMap
-
-        Seq(
-          "ss_sold_date_sk",
-          "ss_item_sk",
-          "ss_ticket_number",
-          "ss_quantity",
-          "ss_wholesale_cost",
-          "ss_list_price",
-          "ss_sales_price",
-          "ss_ext_sales_price",
-          "ss_net_profit"
-        ).foreach {
-          colName =>
-            val row = statsByColumn(colName)
-
-            assert(
-              Option(row.getAs[Any]("min")).exists(_.toString.nonEmpty),
-              s"Expected non-empty min for $colName in SHOW STATS output: $row")
-
-            assert(
-              Option(row.getAs[Any]("max")).exists(_.toString.nonEmpty),
-              s"Expected non-empty max for $colName in SHOW STATS output: $row")
-        }
-      }
-    }
-  }
 }

cpp/velox/CMakeLists.txt

@@ -223,10 +223,8 @@ if(ENABLE_GPU)
     memory/GpuBufferColumnarBatch.cc)
 endif()
 
-if(ENABLE_ENHANCED_FEATURES)
-  list(APPEND VELOX_SRCS compute/iceberg/IcebergFormat.cc
-       compute/iceberg/IcebergWriter.cc)
-endif()
+list(APPEND VELOX_SRCS compute/iceberg/IcebergFormat.cc
+     compute/iceberg/IcebergWriter.cc)
 
 if(BUILD_TESTS OR BUILD_BENCHMARKS)
   set(BUILD_TEST_UTILS ON)

cpp/velox/compute/VeloxBackend.cc

@@ -51,6 +51,7 @@
 #include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveDataSource.h"
+#include "velox/connectors/hive/iceberg/IcebergConnector.h"
 #include "velox/connectors/hive/storage_adapters/abfs/RegisterAbfsFileSystem.h" // @manual
 #include "velox/connectors/hive/storage_adapters/gcs/RegisterGcsFileSystem.h" // @manual
 #include "velox/connectors/hive/storage_adapters/hdfs/HdfsFileSystem.h"
@@ -323,6 +324,13 @@ std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createHiveC
   return std::make_shared<velox::connector::hive::HiveConnector>(connectorId, hiveConnectorConfig_, ioExecutor);
 }
 
+std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createIcebergConnector(
+    const std::string& connectorId,
+    folly::Executor* ioExecutor) const {
+  return std::make_shared<velox::connector::hive::iceberg::IcebergConnector>(
+      connectorId, hiveConnectorConfig_, ioExecutor);
+}
+
 std::shared_ptr<facebook::velox::connector::Connector> VeloxBackend::createValueStreamConnector(
     const std::string& connectorId,
     bool dynamicFilterEnabled) const {

cpp/velox/compute/VeloxBackend.h

@@ -74,6 +74,10 @@ class VeloxBackend {
       const std::string& connectorId,
       folly::Executor* ioExecutor) const;
 
+  std::shared_ptr<facebook::velox::connector::Connector> createIcebergConnector(
+      const std::string& connectorId,
+      folly::Executor* ioExecutor) const;
+
   std::shared_ptr<facebook::velox::connector::Connector> createValueStreamConnector(
       const std::string& connectorId,
       bool dynamicFilterEnabled) const;

cpp/velox/compute/VeloxConnectorIds.h

@@ -23,9 +23,11 @@ namespace gluten {
 
 struct VeloxConnectorIds {
   std::string hive;
+  std::string iceberg;
   std::string iterator;
   std::string cudfHive;
   bool hiveRegistered{false};
+  bool icebergRegistered{false};
   bool iteratorRegistered{false};
   bool cudfHiveRegistered{false};
 };

cpp/velox/compute/VeloxRuntime.cc

@@ -213,6 +213,7 @@ std::string makeScopedConnectorId(const std::string& base, uint64_t runtimeId) {
 VeloxConnectorIds makeScopedConnectorIds(uint64_t runtimeId) {
   return VeloxConnectorIds{
       .hive = makeScopedConnectorId(kHiveConnectorId, runtimeId),
+      .iceberg = makeScopedConnectorId(kIcebergConnectorId, runtimeId),
       .iterator = makeScopedConnectorId(kIteratorConnectorId, runtimeId),
       .cudfHive = makeScopedConnectorId(kCudfHiveConnectorId, runtimeId)};
 }
@@ -271,6 +272,14 @@ void VeloxRuntime::registerConnectors() {
       velox::connector::hasConnector(connectorIds_.hive),
       "Scoped hive connector not found after registration: " + connectorIds_.hive);
 
+  connectorIds_.icebergRegistered =
+      velox::connector::registerConnector(backend->createIcebergConnector(connectorIds_.iceberg, ioExecutor_.get()));
+  GLUTEN_CHECK(
+      connectorIds_.icebergRegistered, "Failed to register scoped Iceberg connector: " + connectorIds_.iceberg);
+  GLUTEN_CHECK(
+      velox::connector::hasConnector(connectorIds_.iceberg),
+      "Scoped Iceberg connector not found after registration: " + connectorIds_.iceberg);
+
   const auto valueStreamDynamicFilterEnabled =
       veloxCfg_->get<bool>(kValueStreamDynamicFilterEnabled, kValueStreamDynamicFilterEnabledDefault);
   connectorIds_.iteratorRegistered = velox::connector::registerConnector(
@@ -310,6 +319,10 @@ void VeloxRuntime::unregisterConnectors() {
     velox::connector::unregisterConnector(connectorIds_.hive);
     connectorIds_.hiveRegistered = false;
   }
+  if (connectorIds_.icebergRegistered) {
+    velox::connector::unregisterConnector(connectorIds_.iceberg);
+    connectorIds_.icebergRegistered = false;
+  }
 }
 
 void VeloxRuntime::parsePlan(const uint8_t* data, int32_t size) {
@@ -488,7 +501,6 @@ std::shared_ptr<RowToColumnarConverter> VeloxRuntime::createRow2ColumnarConverte
   return std::make_shared<VeloxRowToColumnarConverter>(cSchema, veloxPool);
 }
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 std::shared_ptr<IcebergWriter> VeloxRuntime::createIcebergWriter(
     RowTypePtr rowType,
     int32_t format,
@@ -516,7 +528,6 @@ std::shared_ptr<IcebergWriter> VeloxRuntime::createIcebergWriter(
       veloxPool,
       connectorPool);
 }
-#endif
 
 std::shared_ptr<ShuffleWriter> VeloxRuntime::createShuffleWriter(
     int32_t numPartitions,

cpp/velox/compute/VeloxRuntime.h

@@ -20,20 +20,15 @@
 #include "WholeStageResultIterator.h"
 #include "compute/Runtime.h"
 #include "compute/VeloxConnectorIds.h"
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 #include "iceberg/IcebergWriter.h"
-#endif
 #include <folly/Executor.h>
 #include "memory/VeloxMemoryManager.h"
 #include "operators/serializer/VeloxColumnarBatchSerializer.h"
 #include "operators/serializer/VeloxColumnarToRowConverter.h"
 #include "operators/writer/VeloxParquetDataSource.h"
 #include "shuffle/ShuffleReader.h"
 #include "shuffle/ShuffleWriter.h"
-
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
 #include "IcebergNestedField.pb.h"
-#endif
 
 namespace gluten {
 
@@ -76,7 +71,6 @@ class VeloxRuntime final : public Runtime {
 
   std::shared_ptr<RowToColumnarConverter> createRow2ColumnarConverter(struct ArrowSchema* cSchema) override;
 
-#ifdef GLUTEN_ENABLE_ENHANCED_FEATURES
   std::shared_ptr<IcebergWriter> createIcebergWriter(
       RowTypePtr rowType,
       int32_t format,
@@ -88,7 +82,6 @@ class VeloxRuntime final : public Runtime {
       std::shared_ptr<const facebook::velox::connector::hive::iceberg::IcebergPartitionSpec> spec,
       const gluten::IcebergNestedField& protoField,
       const std::unordered_map<std::string, std::string>& sparkConfs);
-#endif
 
   std::shared_ptr<ShuffleWriter> createShuffleWriter(
       int numPartitions,

cpp/velox/compute/WholeStageResultIterator.cc

@@ -161,7 +161,7 @@ WholeStageResultIterator::WholeStageResultIterator(
         std::unordered_map<std::string, std::string> customSplitInfo{{"table_format", "hive-iceberg"}};
         auto deleteFiles = icebergSplitInfo->deleteFilesVec[idx];
         split = std::make_shared<velox::connector::hive::iceberg::HiveIcebergSplit>(
-            connectorIds_.hive,
+            connectorIds_.iceberg,
             paths[idx],
             format,
             starts[idx],
@@ -216,6 +216,7 @@ std::shared_ptr<velox::core::QueryCtx> WholeStageResultIterator::createNewVeloxQ
   std::unordered_map<std::string, std::shared_ptr<velox::config::ConfigBase>> connectorConfigs;
   auto hiveSessionConfig = createHiveConnectorSessionConfig(veloxCfg_);
   connectorConfigs[connectorIds_.hive] = hiveSessionConfig;
+  connectorConfigs[connectorIds_.iceberg] = hiveSessionConfig;
   connectorConfigs[connectorIds_.iterator] = hiveSessionConfig;
 #ifdef GLUTEN_ENABLE_GPU
   if (!connectorIds_.cudfHive.empty()) {

cpp/velox/compute/iceberg/IcebergWriter.cc

@@ -17,8 +17,10 @@
 
 #include "IcebergWriter.h"
 
+#include "IcebergNestedField.pb.h"
 #include "IcebergPartitionSpec.pb.h"
 #include "compute/ProtobufUtils.h"
+#include "compute/VeloxBackend.h"
 #include "compute/iceberg/IcebergFormat.h"
 #include "config/VeloxConfig.h"
 #include "utils/ConfigExtractor.h"
@@ -99,9 +101,9 @@ class GlutenIcebergFileNameGenerator : public connector::hive::FileNameGenerator
   mutable int32_t fileCount_;
 };
 
-iceberg::IcebergNestedField convertToIcebergNestedField(const gluten::IcebergNestedField& protoField) {
-  IcebergNestedField result;
-  result.id = protoField.id();
+parquet::ParquetFieldId convertToIcebergNestedField(const gluten::IcebergNestedField& protoField) {
+  parquet::ParquetFieldId result;
+  result.fieldId = protoField.id();
 
   // Recursively convert children
   result.children.reserve(protoField.children_size());
@@ -121,7 +123,7 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
     int64_t taskId,
     const std::string& operationId,
     std::shared_ptr<const IcebergPartitionSpec> spec,
-    const iceberg::IcebergNestedField& nestedField,
+    const parquet::ParquetFieldId& nestedField,
     facebook::velox::memory::MemoryPool* pool) {
   std::vector<std::shared_ptr<const iceberg::IcebergColumnHandle>> columnHandles;
 
@@ -139,14 +141,12 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kPartitionKey,
           columnTypes.at(i),
-          columnTypes.at(i),
           nestedField.children[i]));
     } else {
       columnHandles.push_back(std::make_shared<iceberg::IcebergColumnHandle>(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kRegular,
           columnTypes.at(i),
-          columnTypes.at(i),
           nestedField.children[i]));
     }
   }
@@ -157,18 +157,9 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
   std::shared_ptr<const connector::hive::LocationHandle> locationHandle =
       std::make_shared<connector::hive::LocationHandle>(
           outputDirectoryPath, outputDirectoryPath, connector::hive::LocationHandle::TableType::kExisting);
-  const std::vector<IcebergSortingColumn> sortedBy;
   const std::unordered_map<std::string, std::string> serdeParameters;
   return std::make_shared<connector::hive::iceberg::IcebergInsertTableHandle>(
-      columnHandles,
-      locationHandle,
-      spec,
-      pool,
-      fileFormat,
-      sortedBy,
-      compressionKind,
-      serdeParameters,
-      fileNameGenerator);
+      columnHandles, locationHandle, fileFormat, spec, compressionKind, serdeParameters, fileNameGenerator);
 }
 
 } // namespace
@@ -200,20 +191,34 @@ IcebergWriter::IcebergWriter(
   connectorSessionProperties_ = createHiveConnectorSessionConfig(veloxCfg);
   connectorConfig_ =
       std::make_shared<facebook::velox::connector::hive::HiveConfig>(createHiveConnectorConfig(veloxCfg));
+  std::unordered_map<std::string, std::shared_ptr<facebook::velox::config::ConfigBase>> connectorConfigs;
+  connectorConfigs[kHiveConnectorId] = connectorSessionProperties_;
+  queryCtx_ = facebook::velox::core::QueryCtx::create(
+      nullptr,
+      facebook::velox::core::QueryConfig{sparkConfs},
+      connectorConfigs,
+      nullptr, // cache
+      pool_,
+      nullptr, // spillExecutor
+      "IcebergWriter");
+
+  auto expressionEvaluator =
+      std::make_unique<facebook::velox::exec::SimpleExpressionEvaluator>(queryCtx_.get(), pool_.get());
+
   connectorQueryCtx_ = std::make_unique<connector::ConnectorQueryCtx>(
       pool_.get(),
       connectorPool_.get(),
       connectorSessionProperties_.get(),
       nullptr,
       common::PrefixSortConfig(),
-      nullptr,
+      std::move(expressionEvaluator),
       nullptr,
       "query.IcebergDataSink",
       "task.IcebergDataSink",
       "planNodeId.IcebergDataSink",
       0,
       "");
-
+  auto icebergConfig = std::make_shared<facebook::velox::connector::hive::iceberg::IcebergConfig>(veloxCfg);
   dataSink_ = std::make_unique<IcebergDataSink>(
       rowType_,
       createIcebergInsertTableHandle(
@@ -229,24 +234,27 @@ IcebergWriter::IcebergWriter(
           pool_.get()),
       connectorQueryCtx_.get(),
       facebook::velox::connector::CommitStrategy::kNoCommit,
-      connectorConfig_);
+      connectorConfig_,
+      icebergConfig);
 }
 
 void IcebergWriter::write(const VeloxColumnarBatch& batch) {
   auto inputRowVector = batch.getRowVector();
   auto inputRowType = asRowType(inputRowVector->type());
 
+  std::vector<VectorPtr> dataColumns;
+
   if (inputRowType->size() != rowType_->size()) {
     const auto& children = inputRowVector->children();
-    std::vector<VectorPtr> dataColumns(children.begin() + 1, children.begin() + 1 + rowType_->size());
-
-    auto filteredRowVector = std::make_shared<RowVector>(
-        pool_.get(), rowType_, inputRowVector->nulls(), inputRowVector->size(), std::move(dataColumns));
-
-    dataSink_->appendData(filteredRowVector);
+    dataColumns.assign(children.begin() + 1, children.begin() + 1 + rowType_->size());
   } else {
-    dataSink_->appendData(inputRowVector);
+    dataColumns = inputRowVector->children();
   }
+
+  auto normalizedRowVector = std::make_shared<RowVector>(
+      pool_.get(), rowType_, inputRowVector->nulls(), inputRowVector->size(), std::move(dataColumns));
+
+  dataSink_->appendData(normalizedRowVector);
 }
 
 std::vector<std::string> IcebergWriter::commit() {