fix

fix

cpp/core/CMakeLists.txt

@@ -209,7 +209,6 @@ set(SPARK_COLUMNAR_PLUGIN_SRCS
         shuffle/rss/CelebornPartitionWriter.cc
         shuffle/Utils.cc
         utils/Compression.cc
-        utils/DebugOut.cc
         utils/StringUtil.cc
         utils/ObjectStore.cc
         jni/JniError.cc

cpp/core/benchmarks/CompressionBenchmark.cc

@@ -25,6 +25,7 @@
 #include <arrow/type_fwd.h>
 #include <benchmark/benchmark.h>
 #include <execinfo.h>
+#include <glog/logging.h>
 #include <parquet/arrow/reader.h>
 #include <parquet/file_reader.h>
 #include <sched.h>
@@ -126,7 +127,7 @@ class BenchmarkCompression {
       }
       case gluten::kQatZstd: {
         ipcWriteOptions.codec = createArrowIpcCodec(arrow::Compression::ZSTD, CodecBackend::QAT);
-        std::cout << "load qatzstd" << std::endl;
+        LOG(INFO) << "load qatzstd";
         break;
       }
 #endif
@@ -163,7 +164,7 @@ class BenchmarkCompression {
         state);
     auto endTime = std::chrono::steady_clock::now();
     auto totalTime = (endTime - startTime).count();
-    std::cout << "Thread " << state.thread_index() << " took " << (1.0 * totalTime / 1e9) << "s" << std::endl;
+    LOG(INFO) << "Thread " << state.thread_index() << " took " << (1.0 * totalTime / 1e9) << "s";
 
     state.counters["rowgroups"] =
         benchmark::Counter(rowGroupIndices_.size(), benchmark::Counter::kAvgThreads, benchmark::Counter::OneK::kIs1000);
@@ -301,8 +302,8 @@ class BenchmarkCompressionCacheScanBenchmark final : public BenchmarkCompression
       }
     } while (recordBatch);
 
-    std::cout << "parquet parse done elapsed time " << elapseRead / 1e6 << " ms " << std::endl;
-    std::cout << "batches = " << numBatches << " rows = " << numRows << std::endl;
+    LOG(INFO) << "parquet parse done elapsed time " << elapseRead / 1e6 << " ms ";
+    LOG(INFO) << "batches = " << numBatches << " rows = " << numRows;
 
     std::vector<std::shared_ptr<arrow::ipc::IpcPayload>> payloads(batches.size());
     std::vector<std::vector<int64_t>> uncompressedBufferSize(batches.size());
@@ -418,26 +419,26 @@ int main(int argc, char** argv) {
     } else if (strcmp(argv[i], "--file") == 0) {
       datafile = argv[i + 1];
     } else if (strcmp(argv[i], "--qat-gzip") == 0) {
-      std::cout << "QAT gzip is used as codec" << std::endl;
+      LOG(INFO) << "QAT gzip is used as codec";
       codec = gluten::kQatGzip;
     } else if (strcmp(argv[i], "--qat-zstd") == 0) {
-      std::cout << "QAT zstd is used as codec" << std::endl;
+      LOG(INFO) << "QAT zstd is used as codec";
       codec = gluten::kQatZstd;
     } else if (strcmp(argv[i], "--qpl-gzip") == 0) {
-      std::cout << "QPL gzip is used as codec" << std::endl;
+      LOG(INFO) << "QPL gzip is used as codec";
       codec = gluten::kQplGzip;
     } else if (strcmp(argv[i], "--zstd") == 0) {
-      std::cout << "CPU zstd is used as codec" << std::endl;
+      LOG(INFO) << "CPU zstd is used as codec";
       codec = gluten::kZstd;
     } else if (strcmp(argv[i], "--buffer-size") == 0) {
       compressBufferSize = atol(argv[i + 1]);
     } else if (strcmp(argv[i], "--cpu-offset") == 0) {
       cpuOffset = atol(argv[i + 1]);
     }
   }
-  std::cout << "iterations = " << iterations << std::endl;
-  std::cout << "threads = " << threads << std::endl;
-  std::cout << "datafile = " << datafile << std::endl;
+  LOG(INFO) << "iterations = " << iterations;
+  LOG(INFO) << "threads = " << threads;
+  LOG(INFO) << "datafile = " << datafile;
 
   gluten::BenchmarkCompressionIterateScanBenchmark bmIterateScan(datafile, compressBufferSize);
   gluten::BenchmarkCompressionCacheScanBenchmark bmCacheScan(datafile, compressBufferSize);

cpp/core/compute/Runtime.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <glog/logging.h>
+
 #include "compute/ProtobufUtils.h"
 #include "compute/ResultIterator.h"
 #include "memory/ArrowMemoryPool.h"
@@ -29,7 +31,6 @@
 #include "shuffle/ShuffleReader.h"
 #include "shuffle/ShuffleWriter.h"
 #include "substrait/plan.pb.h"
-#include "utils/DebugOut.h"
 #include "utils/ObjectStore.h"
 
 namespace gluten {

cpp/core/jni/JniCommon.h

@@ -28,7 +28,6 @@
 #include "memory/AllocationListener.h"
 #include "shuffle/rss/RssClient.h"
 #include "utils/Compression.h"
-#include "utils/DebugOut.h"
 #include "utils/exception.h"
 
 static jint jniVersion = JNI_VERSION_1_8;
@@ -52,7 +51,7 @@ static inline void checkException(JNIEnv* env) {
     std::string description =
         jStringToCString(env, (jstring)env->CallStaticObjectMethod(describerClass, describeMethod, t));
     if (env->ExceptionCheck()) {
-      std::cerr << "Fatal: Uncaught Java exception during calling the Java exception describer method! " << std::endl;
+      LOG(WARNING) << "Fatal: Uncaught Java exception during calling the Java exception describer method! ";
     }
     throw gluten::GlutenException("Error during calling Java code from native code: " + description);
   }
@@ -106,13 +105,11 @@ static inline jmethodID getStaticMethodIdOrError(JNIEnv* env, jclass thisClass,
 static inline void attachCurrentThreadAsDaemonOrThrow(JavaVM* vm, JNIEnv** out) {
   int getEnvStat = vm->GetEnv(reinterpret_cast<void**>(out), jniVersion);
   if (getEnvStat == JNI_EDETACHED) {
-    DEBUG_OUT << "JNIEnv was not attached to current thread." << std::endl;
     // Reattach current thread to JVM
     getEnvStat = vm->AttachCurrentThreadAsDaemon(reinterpret_cast<void**>(out), NULL);
     if (getEnvStat != JNI_OK) {
       throw gluten::GlutenException("Failed to reattach current thread to JVM.");
     }
-    DEBUG_OUT << "Succeeded attaching current thread." << std::endl;
     return;
   }
   if (getEnvStat != JNI_OK) {
@@ -161,7 +158,7 @@ static inline void backtrace() {
   auto size = backtrace(array, 1024);
   char** strings = backtrace_symbols(array, size);
   for (size_t i = 0; i < size; ++i) {
-    std::cout << strings[i] << std::endl;
+    LOG(INFO) << strings[i];
   }
   free(strings);
 }
@@ -229,8 +226,8 @@ class SparkAllocationListener final : public gluten::AllocationListener {
   ~SparkAllocationListener() override {
     JNIEnv* env;
     if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
-      std::cerr << "SparkAllocationListener#~SparkAllocationListener(): "
-                << "JNIEnv was not attached to current thread" << std::endl;
+      LOG(WARNING) << "SparkAllocationListener#~SparkAllocationListener(): "
+                   << "JNIEnv was not attached to current thread";
       return;
     }
     env->DeleteGlobalRef(jListenerGlobalRef_);
@@ -329,8 +326,8 @@ class CelebornClient : public RssClient {
   ~CelebornClient() {
     JNIEnv* env;
     if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
-      std::cerr << "CelebornClient#~CelebornClient(): "
-                << "JNIEnv was not attached to current thread" << std::endl;
+      LOG(WARNING) << "CelebornClient#~CelebornClient(): "
+                   << "JNIEnv was not attached to current thread";
       return;
     }
     env->DeleteGlobalRef(javaCelebornShuffleWriter_);

cpp/core/jni/JniWrapper.cc

@@ -18,7 +18,6 @@
 #include <jni.h>
 #include <filesystem>
 
-#include <glog/logging.h>
 #include "compute/ProtobufUtils.h"
 #include "compute/Runtime.h"
 #include "config/GlutenConfig.h"
@@ -92,11 +91,10 @@ class JavaInputStreamAdaptor final : public arrow::io::InputStream {
     try {
       auto status = JavaInputStreamAdaptor::Close();
       if (!status.ok()) {
-        DEBUG_OUT << __func__ << " call JavaInputStreamAdaptor::Close() failed, status:" << status.ToString()
-                  << std::endl;
+        LOG(WARNING) << __func__ << " call JavaInputStreamAdaptor::Close() failed, status:" << status.ToString();
       }
     } catch (std::exception& e) {
-      DEBUG_OUT << __func__ << " call JavaInputStreamAdaptor::Close() got exception:" << e.what() << std::endl;
+      LOG(WARNING) << __func__ << " call JavaInputStreamAdaptor::Close() got exception:" << e.what();
     }
   }
 
@@ -820,7 +818,7 @@ JNIEXPORT jlong JNICALL Java_io_glutenproject_vectorized_ShuffleWriterJniWrapper
   jobject thread = env->CallStaticObjectMethod(cls, mid);
   checkException(env);
   if (thread == NULL) {
-    std::cerr << "Thread.currentThread() return NULL" << std::endl;
+    LOG(WARNING) << "Thread.currentThread() return NULL";
   } else {
     jmethodID midGetid = getMethodIdOrError(env, cls, "getId", "()J");
     jlong sid = env->CallLongMethod(thread, midGetid);

cpp/core/shuffle/LocalPartitionWriter.cc

@@ -19,7 +19,6 @@
 #include <random>
 #include <thread>
 #include "shuffle/Utils.h"
-#include "utils/DebugOut.h"
 #include "utils/StringUtil.h"
 #include "utils/Timer.h"
 
@@ -129,7 +128,6 @@ class FlushOnSpillEvictHandle final : public LocalPartitionWriter::LocalEvictHan
     ARROW_ASSIGN_OR_RAISE(auto start, os_->Tell());
     RETURN_NOT_OK(arrow::ipc::WriteIpcPayload(*payload, options_, os_.get(), &metadataLength));
     ARROW_ASSIGN_OR_RAISE(auto end, os_->Tell());
-    DEBUG_OUT << "Spilled partition " << partitionId << " file start: " << start << ", file end: " << end << std::endl;
     spillInfo_->partitionSpillInfos.push_back({partitionId, end - start});
     return arrow::Status::OK();
   }

cpp/core/tests/RoundRobinPartitionerTest.cc

@@ -15,6 +15,7 @@
  * limitations under the License.
  */
 #include "shuffle/RoundRobinPartitioner.h"
+#include <glog/logging.h>
 #include <gtest/gtest.h>
 #include <numeric>
 
@@ -46,10 +47,11 @@ class RoundRobinPartitionerTest : public ::testing::Test {
 
   template <typename T>
   void toString(const std::vector<T>& vec, const std::string& name) const {
-    std::cout << name << " = [";
-    std::copy(vec.cbegin(), vec.cend(), std::ostream_iterator<T>(std::cout, ","));
-    std::cout << " ]";
-    std::cout << std::endl;
+    std::stringstream ss;
+    ss << name << " = [";
+    std::copy(vec.cbegin(), vec.cend(), std::ostream_iterator<T>(ss, ","));
+    ss << " ]";
+    LOG(INFO) << ss.str();
   }
 
   int32_t getPidSelection() const {

cpp/core/utils/ObjectStore.cc

@@ -16,6 +16,7 @@
  */
 
 #include "ObjectStore.h"
+#include <glog/logging.h>
 #include <iostream>
 
 gluten::ObjectStore::~ObjectStore() {
@@ -24,7 +25,7 @@ gluten::ObjectStore::~ObjectStore() {
   for (auto itr = aliveObjectHandles_.rbegin(); itr != aliveObjectHandles_.rend(); itr++) {
     ResourceHandle handle = *itr;
     if (store_.lookup(handle) == nullptr) {
-      std::cerr << "Fatal: resource handle " + std::to_string(handle) + " not found in store." << std::endl;
+      LOG(WARNING) << "Fatal: resource handle " + std::to_string(handle) + " not found in store.";
     }
     store_.erase(handle);
   }

cpp/core/utils/Print.h

@@ -19,7 +19,6 @@
 #include <iostream>
 #include <string>
 #include <vector>
-#include "DebugOut.h"
 
 namespace gluten {
 

cpp/core/utils/macros.h

@@ -20,6 +20,7 @@
 #include <time.h>
 
 #include <arrow/status.h>
+#include <glog/logging.h>
 #include <chrono>
 
 #include "utils/exception.h"

cpp/core/utils/qpl/qpl_job_pool.cc

@@ -37,7 +37,7 @@ QplJobHWPool& QplJobHWPool::GetInstance() {
 QplJobHWPool::QplJobHWPool() : randomEngine(std::random_device()()), distribution(0, MAX_JOB_NUMBER - 1) {
   uint64_t initTime = 0;
   TIME_NANO(initTime, InitJobPool());
-  DEBUG_OUT << "Init job pool took " << 1.0 * initTime / 1e6 << "ms" << std::endl;
+  LOG(INFO) << "Init job pool took " << 1.0 * initTime / 1e6 << "ms";
 }
 
 QplJobHWPool::~QplJobHWPool() {
@@ -96,7 +96,7 @@ qpl_job* QplJobHWPool::AcquireJob(uint32_t& jobId) {
     }
   }
   jobId = MAX_JOB_NUMBER - index;
-  DEBUG_OUT << "Acquired job index " << index << " after " << retry << " retries." << std::endl;
+  LOG(INFO) << "Acquired job index " << index << " after " << retry << " retries.";
   return jobPool[index];
 }
 

cpp/velox/benchmarks/ColumnarToRowBenchmark.cc

@@ -131,7 +131,7 @@ class GoogleBenchmarkColumnarToRowCacheScanBenchmark : public GoogleBenchmarkCol
     localSchema = std::make_shared<arrow::Schema>(*schema_.get());
 
     if (state.thread_index() == 0)
-      std::cout << localSchema->ToString() << std::endl;
+      LOG(INFO) << localSchema->ToString();
 
     std::unique_ptr<::parquet::arrow::FileReader> parquetReader;
     std::shared_ptr<RecordBatchReader> recordBatchReader;
@@ -150,7 +150,7 @@ class GoogleBenchmarkColumnarToRowCacheScanBenchmark : public GoogleBenchmarkCol
       }
     } while (recordBatch);
 
-    std::cout << " parquet parse done elapsed time = " << elapseRead / 1000000 << " rows = " << numRows << std::endl;
+    LOG(INFO) << " parquet parse done elapsed time = " << elapseRead / 1000000 << " rows = " << numRows;
 
     // reuse the columnarToRowConverter for batches caused system % increase a lot
     auto ctxPool = defaultLeafVeloxMemoryPool();
@@ -263,10 +263,10 @@ int main(int argc, char** argv) {
       cpu = atol(argv[i + 1]);
     }
   }
-  std::cout << "iterations = " << iterations << std::endl;
-  std::cout << "threads = " << threads << std::endl;
-  std::cout << "datafile = " << datafile << std::endl;
-  std::cout << "cpu = " << cpu << std::endl;
+  LOG(INFO) << "iterations = " << iterations;
+  LOG(INFO) << "threads = " << threads;
+  LOG(INFO) << "datafile = " << datafile;
+  LOG(INFO) << "cpu = " << cpu;
 
   gluten::GoogleBenchmarkColumnarToRowCacheScanBenchmark bck(datafile);