diff --git a/.github/workflows/velox_docker.yml b/.github/workflows/velox_docker.yml index 6c1be4344c71..013f066d221f 100644 --- a/.github/workflows/velox_docker.yml +++ b/.github/workflows/velox_docker.yml @@ -68,6 +68,16 @@ jobs: - name: Build Gluten Velox third party if: ${{ steps.cache.outputs.cache-hit != 'true' }} run: | + yum update -y && yum install -y java-1.8.0-openjdk-devel wget + export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk + + echo "JAVA_HOME: $JAVA_HOME" + + wget https://archive.apache.org/dist/hadoop/core/hadoop-2.10.1/hadoop-2.10.1.tar.gz + tar xf hadoop-2.10.1.tar.gz -C /usr/local/ + export HADOOP_HOME='/usr/local/hadoop-2.10.1' + echo "HADOOP_HOME: $HADOOP_HOME" + source dev/ci-velox-buildstatic.sh - name: Upload Artifact Native uses: actions/upload-artifact@v2 diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala index 459a7886ea23..f4318812e50d 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala @@ -38,7 +38,9 @@ import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.types.{BinaryType, DateType, Decimal, DecimalType, StructType, TimestampType} import org.apache.spark.sql.utils.OASPackageBridge.InputMetricsWrapper import org.apache.spark.sql.vectorized.ColumnarBatch -import org.apache.spark.util.ExecutorManager +import org.apache.spark.util.{ExecutorManager, SerializableConfiguration} + +import org.apache.hadoop.fs.{FileSystem, Path} import java.lang.{Long => JLong} import java.nio.charset.StandardCharsets @@ -53,7 +55,8 @@ class VeloxIteratorApi extends IteratorApi with Logging { partition: InputPartition, partitionSchema: StructType, fileFormat: ReadFileFormat, - metadataColumnNames: Seq[String]): SplitInfo = { + metadataColumnNames: Seq[String], + serializableHadoopConf: SerializableConfiguration): SplitInfo = { partition match { case f: FilePartition => val ( @@ -64,7 +67,7 @@ class VeloxIteratorApi extends IteratorApi with Logging { modificationTimes, partitionColumns, metadataColumns) = - constructSplitInfo(partitionSchema, f.files, metadataColumnNames) + constructSplitInfo(partitionSchema, f.files, metadataColumnNames, serializableHadoopConf) val preferredLocations = SoftAffinity.getFilePartitionLocations(f) LocalFilesBuilder.makeLocalFiles( @@ -105,7 +108,8 @@ class VeloxIteratorApi extends IteratorApi with Logging { private def constructSplitInfo( schema: StructType, files: Array[PartitionedFile], - metadataColumnNames: Seq[String]) = { + metadataColumnNames: Seq[String], + serializableHadoopConf: SerializableConfiguration) = { val paths = new JArrayList[String]() val starts = new JArrayList[JLong] val lengths = new JArrayList[JLong]() @@ -117,9 +121,15 @@ class VeloxIteratorApi extends IteratorApi with Logging { file => // The "file.filePath" in PartitionedFile is not the original encoded path, so the decoded // path is incorrect in some cases and here fix the case of ' ' by using GlutenURLDecoder + var filePath = file.filePath.toString + if (filePath.startsWith("viewfs")) { + val viewPath = new Path(filePath) + val viewFileSystem = FileSystem.get(viewPath.toUri, serializableHadoopConf.value) + filePath = viewFileSystem.resolvePath(viewPath).toString + } paths.add( GlutenURLDecoder - .decode(file.filePath.toString, StandardCharsets.UTF_8.name())) + .decode(filePath, StandardCharsets.UTF_8.name())) starts.add(JLong.valueOf(file.start)) lengths.add(JLong.valueOf(file.length)) val (fileSize, modificationTime) = diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos7.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos7.scala index 47ed2c47cbb5..d77bb145e149 100755 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos7.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos7.scala @@ -36,7 +36,7 @@ class SharedLibraryLoaderCentos7 extends SharedLibraryLoader { .loadAndCreateLink("libntlm.so.0", "libntlm.so", false) .loadAndCreateLink("libgsasl.so.7", "libgsasl.so", false) .loadAndCreateLink("libprotobuf.so.32", "libprotobuf.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .loadAndCreateLink("libre2.so.10", "libre2.so", false) .loadAndCreateLink("libzstd.so.1", "libzstd.so", false) .loadAndCreateLink("liblz4.so.1", "liblz4.so", false) diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos8.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos8.scala index c1d3bf2e26cb..cf7e01d329fd 100755 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos8.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderCentos8.scala @@ -41,7 +41,7 @@ class SharedLibraryLoaderCentos8 extends SharedLibraryLoader { .loadAndCreateLink("libntlm.so.0", "libntlm.so", false) .loadAndCreateLink("libgsasl.so.7", "libgsasl.so", false) .loadAndCreateLink("libprotobuf.so.32", "libprotobuf.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .loadAndCreateLink("libre2.so.0", "libre2.so", false) .loadAndCreateLink("libsodium.so.23", "libsodium.so", false) .commit() diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian11.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian11.scala index ca7d1d22d984..514d84ad0b53 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian11.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian11.scala @@ -46,7 +46,7 @@ class SharedLibraryLoaderDebian11 extends SharedLibraryLoader { .loadAndCreateLink("libsnappy.so.1", "libsnappy.so", false) .loadAndCreateLink("libcurl.so.4", "libcurl.so", false) .loadAndCreateLink("libprotobuf.so.32", "libprotobuf.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .commit() } } diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian12.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian12.scala index 128c8eaa2aef..abd82f4bdbf8 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian12.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderDebian12.scala @@ -52,7 +52,7 @@ class SharedLibraryLoaderDebian12 extends SharedLibraryLoader { .loadAndCreateLink("libevent-2.1.so.7", "libevent-2.1.so", false) .loadAndCreateLink("libcurl.so.4", "libcurl.so", false) .loadAndCreateLink("libprotobuf.so.32", "libprotobuf.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .commit() } } diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2004.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2004.scala index 18f2e6cfbeb3..e0985e11589b 100755 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2004.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2004.scala @@ -59,7 +59,7 @@ class SharedLibraryLoaderUbuntu2004 extends SharedLibraryLoader { .loadAndCreateLink("libicudata.so.66", "libicudata.so", false) .loadAndCreateLink("libicuuc.so.66", "libicuuc.so", false) .loadAndCreateLink("libxml2.so.2", "libxml2.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .loadAndCreateLink("libre2.so.5", "libre2.so", false) .loadAndCreateLink("libsnappy.so.1", "libsnappy.so", false) .loadAndCreateLink("libthrift-0.13.0.so", "libthrift.so", false) diff --git a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2204.scala b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2204.scala index b23105b7dce0..58569f125f39 100755 --- a/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2204.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/utils/SharedLibraryLoaderUbuntu2204.scala @@ -44,7 +44,7 @@ class SharedLibraryLoaderUbuntu2204 extends SharedLibraryLoader { .loadAndCreateLink("libgsasl.so.7", "libgsasl.so", false) .loadAndCreateLink("libprotobuf.so.32", "libprotobuf.so", false) .loadAndCreateLink("libxml2.so.2", "libxml2.so", false) - .loadAndCreateLink("libhdfs3.so.1", "libhdfs3.so", false) + .loadAndCreateLink("libhdfs.so.0.0.0", "libhdfs.so", false) .loadAndCreateLink("libre2.so.9", "libre2.so", false) .loadAndCreateLink("libsnappy.so.1", "libsnappy.so", false) .loadAndCreateLink("libthrift-0.16.0.so", "libthrift.so", false) diff --git a/cpp/velox/CMakeLists.txt b/cpp/velox/CMakeLists.txt index 6d66ea506a7e..289a5b4734cf 100644 --- a/cpp/velox/CMakeLists.txt +++ b/cpp/velox/CMakeLists.txt @@ -531,9 +531,9 @@ set(VELOX_SRCS utils/Common.cc utils/VeloxBatchAppender.cc) -if(ENABLE_HDFS) - list(APPEND VELOX_SRCS utils/HdfsUtils.cc) -endif() +# if (ENABLE_HDFS) +# list(APPEND VELOX_SRCS utils/HdfsUtils.cc) +# endif () if(ENABLE_S3) find_package(ZLIB) @@ -629,8 +629,13 @@ endif() if(ENABLE_HDFS) add_definitions(-DENABLE_HDFS) - find_libhdfs3() - target_link_libraries(velox PUBLIC HDFS::hdfs3) + set(HADOOP_HOME $ENV{HADOOP_HOME}) + find_library( + LIBHDFS + NAMES libhdfs.so + HINTS "${HADOOP_HOME}/lib/native" REQUIRED) + # find_libhdfs3() + target_link_libraries(velox PUBLIC ${LIBHDFS}) endif() if(ENABLE_S3) diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 867d347cdc64..983313819cc3 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/cpp/velox/compute/WholeStageResultIterator.cc @@ -22,9 +22,9 @@ #include "velox/connectors/hive/HiveConnectorSplit.h" #include "velox/exec/PlanNodeStats.h" -#ifdef ENABLE_HDFS -#include "utils/HdfsUtils.h" -#endif +// #ifdef ENABLE_HDFS +// #include "utils/HdfsUtils.h" +// #endif using namespace facebook; @@ -68,9 +68,9 @@ WholeStageResultIterator::WholeStageResultIterator( scanNodeIds_(scanNodeIds), scanInfos_(scanInfos), streamIds_(streamIds) { -#ifdef ENABLE_HDFS - gluten::updateHdfsTokens(veloxCfg_.get()); -#endif + // #ifdef ENABLE_HDFS + // gluten::updateHdfsTokens(veloxCfg_.get()); + // #endif spillStrategy_ = veloxCfg_->get(kSpillStrategy, kSpillStrategyDefaultValue); auto spillThreadNum = veloxCfg_->get(kSpillThreadNum, kSpillThreadNumDefaultValue); if (spillThreadNum > 0) { diff --git a/cpp/velox/utils/HdfsUtils.cc b/cpp/velox/utils/HdfsUtils.cc deleted file mode 100644 index a912c04eee7e..000000000000 --- a/cpp/velox/utils/HdfsUtils.cc +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 "HdfsUtils.h" -#include -#include "config/GlutenConfig.h" -#include "utils/exception.h" - -namespace gluten { - -namespace { -struct Credential { - const std::string userName; - const std::string allTokens; - - bool operator==(const Credential& rhs) const { - return userName == rhs.userName && allTokens == rhs.allTokens; - } - bool operator!=(const Credential& rhs) const { - return !(rhs == *this); - } -}; -} // namespace - -void updateHdfsTokens(const facebook::velox::Config* veloxCfg) { - static std::mutex mtx; - std::lock_guard lock{mtx}; - - static std::optional activeCredential{std::nullopt}; - - const auto& newUserName = veloxCfg->get(gluten::kUGIUserName); - const auto& newAllTokens = veloxCfg->get(gluten::kUGITokens); - - if (!newUserName.hasValue() || !newAllTokens.hasValue()) { - return; - } - - Credential newCredential{newUserName.value(), newAllTokens.value()}; - - if (activeCredential.has_value() && activeCredential.value() == newCredential) { - // Do nothing if the credential is the same with before. - return; - } - - hdfsSetDefautUserName(newCredential.userName.c_str()); - std::vector tokens; - folly::split('\0', newCredential.allTokens, tokens); - for (auto& token : tokens) - hdfsSetTokenForDefaultUser(token.data()); - activeCredential.emplace(newCredential); -} -} // namespace gluten diff --git a/cpp/velox/utils/HdfsUtils.h b/cpp/velox/utils/HdfsUtils.h deleted file mode 100644 index cd017f250ad2..000000000000 --- a/cpp/velox/utils/HdfsUtils.h +++ /dev/null @@ -1,22 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 -#include -namespace gluten { -void updateHdfsTokens(const facebook::velox::Config* veloxCfg); -} diff --git a/ep/build-velox/src/get_velox.sh b/ep/build-velox/src/get_velox.sh index 49654d8a8cd5..dafe15e40f4c 100755 --- a/ep/build-velox/src/get_velox.sh +++ b/ep/build-velox/src/get_velox.sh @@ -17,7 +17,7 @@ set -exu VELOX_REPO=https://github.com/oap-project/velox.git -VELOX_BRANCH=2024_06_14 +VELOX_BRANCH=libhdfs VELOX_HOME="" #Set on run gluten on HDFS diff --git a/ep/build-velox/src/modify_velox.patch b/ep/build-velox/src/modify_velox.patch index 81560917d620..1176aaf5520d 100644 --- a/ep/build-velox/src/modify_velox.patch +++ b/ep/build-velox/src/modify_velox.patch @@ -39,27 +39,7 @@ diff --git a/CMakeLists.txt b/CMakeLists.txt index 5c7bf770a..9f897f577 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt -@@ -234,10 +234,15 @@ if(VELOX_ENABLE_ABFS) - endif() - - if(VELOX_ENABLE_HDFS) -- find_library( -- LIBHDFS3 -- NAMES libhdfs3.so libhdfs3.dylib -- HINTS "${CMAKE_SOURCE_DIR}/hawq/depends/libhdfs3/_build/src/" REQUIRED) -+ find_package(libhdfs3) -+ if(libhdfs3_FOUND AND TARGET HDFS::hdfs3) -+ set(LIBHDFS3 HDFS::hdfs3) -+ else() -+ find_library( -+ LIBHDFS3 -+ NAMES libhdfs3.so libhdfs3.dylib -+ HINTS "${CMAKE_SOURCE_DIR}/hawq/depends/libhdfs3/_build/src/" REQUIRED) -+ endif() - add_definitions(-DVELOX_ENABLE_HDFS3) - endif() - -@@ -377,7 +382,7 @@ resolve_dependency(Boost 1.77.0 COMPONENTS ${BOOST_INCLUDE_LIBRARIES}) +@@ -386,7 +391,7 @@ resolve_dependency(Boost 1.77.0 COMPONENTS ${BOOST_INCLUDE_LIBRARIES}) # for reference. find_package(range-v3) set_source(gflags) diff --git a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala index 53dc8f47861f..91e13fcb0f10 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/backendsapi/IteratorApi.scala @@ -29,6 +29,7 @@ import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.types.StructType import org.apache.spark.sql.utils.OASPackageBridge.InputMetricsWrapper import org.apache.spark.sql.vectorized.ColumnarBatch +import org.apache.spark.util.SerializableConfiguration trait IteratorApi { @@ -36,7 +37,8 @@ trait IteratorApi { partition: InputPartition, partitionSchema: StructType, fileFormat: ReadFileFormat, - metadataColumnNames: Seq[String]): SplitInfo + metadataColumnNames: Seq[String], + serializableHadoopConf: SerializableConfiguration): SplitInfo /** Generate native row partition. */ def genPartitions( diff --git a/gluten-core/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala b/gluten-core/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala index 3bbd99c50a6a..509acbef5b35 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala @@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.connector.read.InputPartition import org.apache.spark.sql.hive.HiveTableScanExecTransformer import org.apache.spark.sql.types.{BooleanType, StringType, StructField, StructType} +import org.apache.spark.util.SerializableConfiguration import com.google.protobuf.StringValue @@ -62,14 +63,21 @@ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource def getProperties: Map[String, String] = Map.empty /** Returns the split infos that will be processed by the underlying native engine. */ - def getSplitInfos: Seq[SplitInfo] = { - getSplitInfosFromPartitions(getPartitions) + def getSplitInfos(serializableHadoopConf: SerializableConfiguration): Seq[SplitInfo] = { + getSplitInfosFromPartitions(getPartitions, serializableHadoopConf) } - def getSplitInfosFromPartitions(partitions: Seq[InputPartition]): Seq[SplitInfo] = { + def getSplitInfosFromPartitions( + partitions: Seq[InputPartition], + serializableHadoopConf: SerializableConfiguration): Seq[SplitInfo] = { partitions.map( BackendsApiManager.getIteratorApiInstance - .genSplitInfo(_, getPartitionSchema, fileFormat, getMetadataColumns.map(_.name))) + .genSplitInfo( + _, + getPartitionSchema, + fileFormat, + getMetadataColumns.map(_.name), + serializableHadoopConf)) } override protected def doValidateInternal(): ValidationResult = { diff --git a/gluten-core/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala b/gluten-core/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala index a49e8aa518b6..8e03b03592f0 100644 --- a/gluten-core/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala +++ b/gluten-core/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala @@ -39,6 +39,7 @@ import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.datasources.FilePartition import org.apache.spark.sql.execution.metric.SQLMetric import org.apache.spark.sql.vectorized.ColumnarBatch +import org.apache.spark.util.SerializableConfiguration import com.google.common.collect.Lists @@ -127,6 +128,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f BackendsApiManager.getMetricsApiInstance.genWholeStageTransformerMetrics(sparkContext) val sparkConf: SparkConf = sparkContext.getConf + val serializableHadoopConf: SerializableConfiguration = new SerializableConfiguration( + sparkContext.hadoopConfiguration) val numaBindingInfo: GlutenNumaBindingInfo = GlutenConfig.getConf.numaBindingInfo val substraitPlanLogLevel: String = GlutenConfig.getConf.substraitPlanLogLevel @@ -289,12 +292,16 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f */ val allScanPartitions = basicScanExecTransformers.map(_.getPartitions) val allScanSplitInfos = - getSplitInfosFromPartitions(basicScanExecTransformers, allScanPartitions) + getSplitInfosFromPartitions( + basicScanExecTransformers, + allScanPartitions, + serializableHadoopConf) val inputPartitions = BackendsApiManager.getIteratorApiInstance.genPartitions( wsCtx, allScanSplitInfos, basicScanExecTransformers) + val rdd = new GlutenWholeStageColumnarRDD( sparkContext, inputPartitions, @@ -369,7 +376,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f private def getSplitInfosFromPartitions( basicScanExecTransformers: Seq[BasicScanExecTransformer], - allScanPartitions: Seq[Seq[InputPartition]]): Seq[Seq[SplitInfo]] = { + allScanPartitions: Seq[Seq[InputPartition]], + serializableHadoopConf: SerializableConfiguration): Seq[Seq[SplitInfo]] = { // If these are two scan transformers, they must have same partitions, // otherwise, exchange will be inserted. We should combine the two scan // transformers' partitions with same index, and set them together in @@ -387,7 +395,8 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f // p1n | p2n => substraitContext.setSplitInfo([p1n, p2n]) val allScanSplitInfos = allScanPartitions.zip(basicScanExecTransformers).map { - case (partition, transformer) => transformer.getSplitInfosFromPartitions(partition) + case (partition, transformer) => + transformer.getSplitInfosFromPartitions(partition, serializableHadoopConf) } val partitionLength = allScanSplitInfos.head.size if (allScanSplitInfos.exists(_.size != partitionLength)) {