diff --git a/.github/workflows/velox_backend_enhanced.yml b/.github/workflows/velox_backend_enhanced.yml index 2337993db3b..efd77b101c8 100644 --- a/.github/workflows/velox_backend_enhanced.yml +++ b/.github/workflows/velox_backend_enhanced.yml @@ -127,7 +127,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 -Piceberg-test -Pdelta -Phudi \ + $MVN_CMD clean test -Pspark-3.4 -Pjava-17 -Pbackends-velox -Pdelta -Phudi \ -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.SkipTest \ -DargLine="-Dspark.test.home=$SPARK_HOME" - name: Upload test report @@ -185,7 +185,7 @@ jobs: java -version export SPARK_HOME=/opt/shims/spark35/spark_home/ ls -l $SPARK_HOME - $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi \ + $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Pdelta -Phudi \ -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.SkipTest \ -DargLine="-Dspark.test.home=$SPARK_HOME" - name: Upload test report @@ -237,7 +237,7 @@ jobs: export JAVA_HOME=/usr/lib/jvm/java-17-openjdk export PATH=$JAVA_HOME/bin:$PATH java -version - $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi -Ppaimon -Pspark-ut \ + $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Pdelta -Phudi -Ppaimon -Pspark-ut \ -DargLine="-Dspark.test.home=/opt/shims/spark35/spark_home/" \ -DtagsToInclude=org.apache.spark.tags.ExtendedSQLTest - name: Upload test report diff --git a/.github/workflows/velox_backend_x86.yml b/.github/workflows/velox_backend_x86.yml index 3ac8edc4244..6edd8162ea1 100644 --- a/.github/workflows/velox_backend_x86.yml +++ b/.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 diff --git a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala index de9eaba5979..edb30dac61f 100644 --- a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala +++ b/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 diff --git a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxTPCHIcebergSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxTPCHIcebergSuite.scala index 42820c04dca..5456e0ba89d 100644 --- a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxTPCHIcebergSuite.scala +++ b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxTPCHIcebergSuite.scala @@ -16,8 +16,6 @@ */ package org.apache.gluten.execution -import org.apache.gluten.tags.SkipTest - import org.apache.spark.SparkConf import org.apache.spark.sql.functions.col @@ -25,7 +23,6 @@ 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" diff --git a/cpp/velox/compute/VeloxBackend.cc b/cpp/velox/compute/VeloxBackend.cc index 45a64908e19..2835b465875 100644 --- a/cpp/velox/compute/VeloxBackend.cc +++ b/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(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(kIteratorConnectorId, hiveConf)); diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 157f9829b81..b082ade29be 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -155,18 +155,16 @@ WholeStageResultIterator::WholeStageResultIterator( std::shared_ptr split; if (auto icebergSplitInfo = std::dynamic_pointer_cast(scanInfo)) { - // Set Iceberg split. - std::unordered_map customSplitInfo{{"table_format", "hive-iceberg"}}; auto deleteFiles = icebergSplitInfo->deleteFilesVec[idx]; split = std::make_shared( - kHiveConnectorId, + kIcebergConnectorId, paths[idx], format, starts[idx], lengths[idx], partitionKeys, std::nullopt, - customSplitInfo, + std::unordered_map(), nullptr, true, deleteFiles, @@ -215,6 +213,7 @@ WholeStageResultIterator::WholeStageResultIterator( std::shared_ptr WholeStageResultIterator::createNewVeloxQueryCtx() { std::unordered_map> connectorConfigs; connectorConfigs[kHiveConnectorId] = createHiveConnectorSessionConfig(veloxCfg_); + connectorConfigs[kIcebergConnectorId] = connectorConfigs[kHiveConnectorId]; std::shared_ptr ctx = velox::core::QueryCtx::create( nullptr, facebook::velox::core::QueryConfig{getQueryContextConf()}, diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index 566ce875aac..da7ec2c3ef2 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/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"; diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc b/cpp/velox/substrait/SubstraitToVeloxPlan.cc index 727f4882e17..7ff46182905 100644 --- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc +++ b/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>& splitInfo #endif } +bool isIcebergSplit(const std::vector>& splitInfos) { + if (splitInfos.empty()) { + return false; + } + if (auto icebergSplitInfo = std::dynamic_pointer_cast(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( connectorId, "hive_table", filterPushdownEnabled, std::move(subfieldFilters), remainingFilter, tableSchema);