Merge branch 'main' into rebase_iceberg

Author: zhouyuan

.github/workflows/util/check.sh

@@ -15,13 +15,30 @@
 # limitations under the License.
 
 export BASE_COMMIT=$1
+
+RESULT=0
+
 ./.github/workflows/util/check.py header branch
 if [ $? -ne 0 ]; then
   ./.github/workflows/util/check.py header branch --fix
   echo -e "\n==== Apply using:"
   echo "patch -p1 \<<EOF"
   git --no-pager diff
   echo "EOF"
-  false
+  RESULT=1
+fi
+
+# Check that shell scripts use #!/usr/bin/env bash instead of #!/bin/bash
+BAD_SHEBANGS=$(git diff --relative --name-only --diff-filter='ACM' "$BASE_COMMIT" -- '*.sh' | while read -r f; do
+  [ -f "$f" ] && head -1 "$f" | grep -q '^#!/bin/bash' && echo "$f"
+done)
+
+if [ -n "$BAD_SHEBANGS" ]; then
+  echo -e "\n==== The following scripts use #!/bin/bash instead of #!/usr/bin/env bash:"
+  echo "$BAD_SHEBANGS"
+  echo "Please replace '#!/bin/bash' with '#!/usr/bin/env bash' for portability."
+  RESULT=1
 fi
 
+exit $RESULT
+

.github/workflows/util/install-flink-resources.sh

@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 # 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.

.github/workflows/util/install-resources.sh

@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 # 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.

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala

@@ -551,12 +551,6 @@ object VeloxConfig extends ConfigRegistry {
       .stringConf
       .createWithDefault("legacy")
 
-  val AWS_S3_CONNECT_TIMEOUT =
-    buildConf("spark.gluten.velox.fs.s3a.connect.timeout")
-      .doc("Timeout for AWS s3 connection.")
-      .stringConf
-      .createWithDefault("200s")
-
   val VELOX_ORC_SCAN_ENABLED =
     buildConf("spark.gluten.sql.columnar.backend.velox.orc.scan.enabled")
       .doc("Enable velox orc scan. If disabled, vanilla spark orc scan will be used.")

backends-velox/src/main/scala/org/apache/gluten/execution/ColumnarCollectLimitExec.scala

@@ -19,6 +19,7 @@ package org.apache.gluten.execution
 import org.apache.gluten.columnarbatch.ColumnarBatches
 import org.apache.gluten.columnarbatch.VeloxColumnarBatches
 
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
@@ -96,6 +97,24 @@ case class ColumnarCollectLimitExec(
     }
   }
 
+  override def executeCollect(): Array[InternalRow] = {
+    val rowsRdd = child.executeColumnar().mapPartitions {
+      it =>
+        val rows = VeloxColumnarToRowExec.toRowIterator(it)
+        rows.map(_.copy())
+    }
+
+    if (limit >= 0) {
+      if (offset > 0) {
+        rowsRdd.take(limit).drop(offset)
+      } else {
+        rowsRdd.take(limit)
+      }
+    } else {
+      rowsRdd.collect().drop(offset)
+    }
+  }
+
   override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
     copy(child = newChild)
 }

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxColumnarToRowExec.scala

@@ -92,6 +92,15 @@ case class VeloxColumnarToRowExec(child: SparkPlan) extends ColumnarToRowExecBas
       VeloxColumnarToRowExec.toRowIterator(_, numOutputRows, numInputBatches, convertTime))
   }
 
+  override def executeCollect(): Array[InternalRow] = {
+    child match {
+      case l: ColumnarCollectLimitExec =>
+        l.executeCollect()
+      case _ =>
+        super.executeCollect()
+    }
+  }
+
   protected def withNewChildInternal(newChild: SparkPlan): VeloxColumnarToRowExec =
     copy(child = newChild)
 }

cpp/velox/compute/VeloxBackend.cc

@@ -314,7 +314,7 @@ void VeloxBackend::initConnector(const std::shared_ptr<velox::config::ConfigBase
       ioThreads >= 0,
       kVeloxIOThreads + " was set to negative number " + std::to_string(ioThreads) + ", this should not happen.");
   if (ioThreads > 0) {
-    ioExecutor_ = std::make_unique<folly::IOThreadPoolExecutor>(ioThreads);
+    ioExecutor_ = std::make_unique<folly::CPUThreadPoolExecutor>(ioThreads);
   }
   velox::connector::registerConnector(
       std::make_shared<velox::connector::hive::HiveConnector>(kHiveConnectorId, hiveConf, ioExecutor_.get()));

cpp/velox/compute/VeloxBackend.h

@@ -90,7 +90,7 @@ class VeloxBackend {
   std::shared_ptr<facebook::velox::cache::AsyncDataCache> asyncDataCache_;
 
   std::unique_ptr<folly::IOThreadPoolExecutor> ssdCacheExecutor_;
-  std::unique_ptr<folly::IOThreadPoolExecutor> ioExecutor_;
+  std::unique_ptr<folly::CPUThreadPoolExecutor> ioExecutor_;
   std::shared_ptr<facebook::velox::memory::MmapAllocator> cacheAllocator_;
 
   std::string cachePathPrefix_;

cpp/velox/utils/ConfigExtractor.cc

@@ -63,7 +63,7 @@ void getS3HiveConfig(
       {S3Config::Keys::kPathStyleAccess, std::make_pair("path.style.access", "false")},
       {S3Config::Keys::kMaxAttempts, std::make_pair("retry.limit", std::nullopt)},
       {S3Config::Keys::kRetryMode, std::make_pair("retry.mode", "legacy")},
-      {S3Config::Keys::kMaxConnections, std::make_pair("connection.maximum", "15")},
+      {S3Config::Keys::kMaxConnections, std::make_pair("connection.maximum", "25")},
       {S3Config::Keys::kSocketTimeout, std::make_pair("connection.timeout", "200s")},
       {S3Config::Keys::kConnectTimeout, std::make_pair("connection.establish.timeout", "30s")},
       {S3Config::Keys::kUseInstanceCredentials, std::make_pair("instance.credentials", "false")},

dev/bloop-test.sh

@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/usr/bin/env bash
 # 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.