[GLUTEN-11630][VL] Enable iceberg tests

.github/workflows/velox_backend_x86.yml

@@ -821,7 +821,7 @@ jobs:
           java -version
           export SPARK_HOME=/opt/shims/spark34/spark_home/
           ls -l $SPARK_HOME
-          $MVN_CMD clean test -Pspark-3.4 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi -Ppaimon -Pspark-ut \
+          $MVN_CMD clean test -Pspark-3.4 -Pjava-17 -Pbackends-velox -Piceberg -Piceberg-test -Pdelta -Phudi -Ppaimon -Pspark-ut \
           -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest \
           -DargLine="-Dspark.test.home=$SPARK_HOME"
       - name: Upload test report

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

@@ -16,7 +16,4 @@
  */
 package org.apache.gluten.execution
 
-import org.apache.gluten.tags.SkipTest
-
-@SkipTest
 class VeloxIcebergSuite extends IcebergSuite

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

@@ -16,16 +16,13 @@
  */
 package org.apache.gluten.execution
 
-import org.apache.gluten.tags.SkipTest
-
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.functions.col
 
 import org.apache.iceberg.spark.SparkWriteOptions
 
 import java.io.File
 
-@SkipTest
 class VeloxTPCHIcebergSuite extends VeloxTPCHSuite {
   protected val tpchBasePath: String =
     getClass.getResource("/").getPath + "../../../src/test/resources"

cpp/velox/compute/VeloxBackend.cc

@@ -46,6 +46,7 @@
 #include "velox/common/file/FileSystems.h"
 #include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/connectors/hive/HiveConnector.h"
+#include "velox/connectors/hive/iceberg/IcebergConnector.h"
 #include "velox/connectors/hive/HiveDataSource.h"
 #include "operators/plannodes/RowVectorStream.h"
 #include "velox/connectors/hive/storage_adapters/abfs/RegisterAbfsFileSystem.h" // @manual
@@ -317,7 +318,13 @@ void VeloxBackend::initConnector(const std::shared_ptr<velox::config::ConfigBase
   }
   velox::connector::registerConnector(
       std::make_shared<velox::connector::hive::HiveConnector>(kHiveConnectorId, hiveConf, ioExecutor_.get()));
-
+  IcebergConnectorFactory icebergFactory;
+  auto icebergConnector = icebergFactory.newConnector(
+      kIcebergConnectorId,
+      hiveConf,
+      ioExecutor_.get());
+    connector::registerConnector(icebergConnector);
+
   // Register value-stream connector for runtime iterator-based inputs
   velox::connector::registerConnector(std::make_shared<ValueStreamConnector>(kIteratorConnectorId, hiveConf));
 

cpp/velox/compute/WholeStageResultIterator.cc

@@ -155,18 +155,16 @@ WholeStageResultIterator::WholeStageResultIterator(
 
       std::shared_ptr<velox::connector::ConnectorSplit> split;
       if (auto icebergSplitInfo = std::dynamic_pointer_cast<IcebergSplitInfo>(scanInfo)) {
-        // Set Iceberg split.
-        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>(
-            kHiveConnectorId,
+            kIcebergConnectorId,
             paths[idx],
             format,
             starts[idx],
             lengths[idx],
             partitionKeys,
             std::nullopt,
-            customSplitInfo,
+            std::unordered_map<std::string, std::string>(),
             nullptr,
             true,
             deleteFiles,
@@ -215,6 +213,7 @@ WholeStageResultIterator::WholeStageResultIterator(
 std::shared_ptr<velox::core::QueryCtx> WholeStageResultIterator::createNewVeloxQueryCtx() {
   std::unordered_map<std::string, std::shared_ptr<velox::config::ConfigBase>> connectorConfigs;
   connectorConfigs[kHiveConnectorId] = createHiveConnectorSessionConfig(veloxCfg_);
+  connectorConfigs[kIcebergConnectorId] = connectorConfigs[kHiveConnectorId];
   std::shared_ptr<velox::core::QueryCtx> ctx = velox::core::QueryCtx::create(
       nullptr,
       facebook::velox::core::QueryConfig{getQueryContextConf()},

cpp/velox/config/VeloxConfig.h

@@ -103,6 +103,7 @@ const std::string kVeloxMemReclaimMaxWaitMs = "spark.gluten.sql.columnar.backend
 const uint64_t kVeloxMemReclaimMaxWaitMsDefault = 3600000; // 60min
 
 const std::string kHiveConnectorId = "test-hive";
+const std::string kIcebergConnectorId = "test-iceberg";
 const std::string kVeloxCacheEnabled = "spark.gluten.sql.columnar.backend.velox.cacheEnabled";
 
 const std::string kExprMaxCompiledRegexes = "spark.gluten.sql.columnar.backend.velox.maxCompiledRegexes";

cpp/velox/substrait/SubstraitToVeloxPlan.cc

@@ -27,6 +27,7 @@
 #include "utils/ConfigExtractor.h"
 #include "utils/VeloxWriterUtils.h"
 
+#include "compute/iceberg/IcebergPlanConverter.h"
 #include "config.pb.h"
 #include "config/GlutenConfig.h"
 #include "config/VeloxConfig.h"
@@ -54,6 +55,16 @@ bool useCudfTableHandle(const std::vector<std::shared_ptr<SplitInfo>>& splitInfo
 #endif
 }
 
+bool isIcebergSplit(const std::vector<std::shared_ptr<SplitInfo>>& splitInfos) {
+  if (splitInfos.empty()) {
+    return false;
+  }
+  if (auto icebergSplitInfo = std::dynamic_pointer_cast<IcebergSplitInfo>(splitInfos[0])) {
+    return true;
+  }
+  return false;
+}
+
 core::SortOrder toSortOrder(const ::substrait::SortField& sortField) {
   switch (sortField.direction()) {
     case ::substrait::SortField_SortDirection_SORT_DIRECTION_ASC_NULLS_FIRST:
@@ -1466,6 +1477,10 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
     connectorId = kCudfHiveConnectorId;
 #endif
   }
+  if (isIcebergSplit(splitInfos_)) {
+    connectorId = kIcebergConnectorId;
+  }
+
   common::SubfieldFilters subfieldFilters;
   tableHandle = std::make_shared<connector::hive::HiveTableHandle>(
       connectorId, "hive_table", filterPushdownEnabled, std::move(subfieldFilters), remainingFilter, tableSchema);