Merge pull request #73 from ClickHouse/apply-clickhouse-patches-to-release-23.0.0

Apply clickhouse patches to release 23.0.0

Author: vitlibar

cpp/src/arrow/adapters/orc/adapter.cc

@@ -223,6 +223,10 @@ class ORCFileReader::Impl {
     return Init();
   }
 
+  virtual liborc::Reader* GetRawORCReader() {
+    return reader_.get();
+  }
+
   Status Init() {
     int64_t nstripes = reader_->getNumberOfStripes();
     stripes_.resize(static_cast<size_t>(nstripes));
@@ -569,6 +573,15 @@ Result<std::unique_ptr<ORCFileReader>> ORCFileReader::Open(
   return result;
 }
 
+Status ORCFileReader::Open(const std::shared_ptr<io::RandomAccessFile>& file,
+                           MemoryPool* pool, std::unique_ptr<ORCFileReader>* reader) {
+  return Open(file, pool).Value(reader);
+}
+
+liborc::Reader* ORCFileReader::GetRawORCReader() {
+  return impl_->GetRawORCReader();
+}
+
 Result<std::shared_ptr<const KeyValueMetadata>> ORCFileReader::ReadMetadata() {
   return impl_->ReadMetadata();
 }

cpp/src/arrow/adapters/orc/adapter.h

@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "arrow/adapters/orc/options.h"
+#include "arrow/adapters/orc/util.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
@@ -61,6 +62,19 @@ class ARROW_EXPORT ORCFileReader {
   static Result<std::unique_ptr<ORCFileReader>> Open(
       const std::shared_ptr<io::RandomAccessFile>& file, MemoryPool* pool);
 
+  /// \brief Creates a new ORC reader.
+  ///
+  /// \param[in] file the data source
+  /// \param[in] pool a MemoryPool to use for buffer allocations
+  /// \param[out] reader the returned reader object
+  /// \return Status
+  ARROW_DEPRECATED("Deprecated in 6.0.0. Use Result-returning overload instead.")
+  static Status Open(const std::shared_ptr<io::RandomAccessFile>& file, MemoryPool* pool,
+                     std::unique_ptr<ORCFileReader>* reader);
+
+  /// \brief Get ORC reader from inside.
+  liborc::Reader* GetRawORCReader();
+
   /// \brief Return the schema read from the ORC file
   ///
   /// \return the returned Schema object

cpp/src/arrow/ipc/reader.cc

@@ -1384,6 +1384,23 @@ class RecordBatchFileReaderImpl : public RecordBatchFileReader {
     return total;
   }
 
+  Result<int64_t> RecordBatchCountRows(int i) override {
+    DCHECK_GE(i, 0);
+    DCHECK_LT(i, num_record_batches());
+    ARROW_ASSIGN_OR_RAISE(auto outer_message,
+                          ReadMessageFromBlock(GetRecordBatchBlock(i)));
+    auto metadata = outer_message->metadata();
+    const flatbuf::Message* message = nullptr;
+    RETURN_NOT_OK(
+        internal::VerifyMessage(metadata->data(), metadata->size(), &message));
+    auto batch = message->header_as_RecordBatch();
+    if (batch == nullptr) {
+      return Status::IOError(
+          "Header-type of flatbuffer-encoded Message is not RecordBatch.");
+    }
+    return batch->length();
+  }
+
   Status Open(const std::shared_ptr<io::RandomAccessFile>& file, int64_t footer_offset,
               const IpcReadOptions& options) {
     owned_file_ = file;

cpp/src/arrow/ipc/reader.h

@@ -203,6 +203,8 @@ class ARROW_EXPORT RecordBatchFileReader
   /// \brief Computes the total number of rows in the file.
   virtual Result<int64_t> CountRows() = 0;
 
+  virtual Result<int64_t> RecordBatchCountRows(int i) = 0;
+
   /// \brief Begin loading metadata for the desired batches into memory.
   ///
   /// This method will also begin loading all dictionaries messages into memory.

cpp/src/arrow/table.cc

@@ -654,7 +654,6 @@ TableBatchReader::TableBatchReader(const Table& table)
   for (int i = 0; i < table.num_columns(); ++i) {
     column_data_[i] = table.column(i).get();
   }
-  DCHECK(table_.Validate().ok());
 }
 
 TableBatchReader::TableBatchReader(std::shared_ptr<Table> table)
@@ -668,7 +667,6 @@ TableBatchReader::TableBatchReader(std::shared_ptr<Table> table)
   for (int i = 0; i < owned_table_->num_columns(); ++i) {
     column_data_[i] = owned_table_->column(i).get();
   }
-  DCHECK(table_.Validate().ok());
 }
 
 std::shared_ptr<Schema> TableBatchReader::schema() const { return table_.schema(); }

cpp/src/arrow/type_fwd.h

@@ -60,7 +60,7 @@ using BufferVector = std::vector<std::shared_ptr<Buffer>>;
 
 class DataType;
 class Field;
-class FieldRef;
+class FieldRef; /// NOLINT(bugprone-forward-declaration-namespace)
 class KeyValueMetadata;
 enum class Endianness;
 class Schema;

cpp/src/arrow/util/logging.cc

@@ -24,6 +24,7 @@
 #endif
 #include <cstdlib>
 #include <iostream>
+#include <sstream>
 
 #ifdef ARROW_USE_GLOG
 
@@ -65,33 +66,33 @@ class CerrLog {
  public:
   explicit CerrLog(ArrowLogLevel severity) : severity_(severity), has_logged_(false) {}
 
-  virtual ~CerrLog() {
+  virtual ~CerrLog() noexcept(false) {
     if (has_logged_) {
-      std::cerr << std::endl;
+      stream << std::endl;
     }
     if (severity_ == ArrowLogLevel::ARROW_FATAL) {
-      PrintBackTrace();
-      std::abort();
+      throw std::runtime_error(stream.str());
     }
   }
 
   std::ostream& Stream() {
     has_logged_ = true;
-    return std::cerr;
+    return stream;
   }
 
   template <class T>
   CerrLog& operator<<(const T& t) {
     if (severity_ != ArrowLogLevel::ARROW_DEBUG) {
       has_logged_ = true;
-      std::cerr << t;
+      stream << t;
     }
     return *this;
   }
 
  protected:
   const ArrowLogLevel severity_;
   bool has_logged_;
+  std::stringstream stream;
 
   void PrintBackTrace() {
 #ifdef ARROW_WITH_BACKTRACE
@@ -250,7 +251,7 @@ std::ostream& ArrowLog::Stream() {
 
 bool ArrowLog::IsEnabled() const { return is_enabled_; }
 
-ArrowLog::~ArrowLog() {
+ArrowLog::~ArrowLog() noexcept(false) {
   if (logging_provider_ != nullptr) {
     delete reinterpret_cast<LoggingProvider*>(logging_provider_);
     logging_provider_ = nullptr;

cpp/src/arrow/util/logging.h

@@ -149,7 +149,7 @@ enum class ArrowLogLevel : int {
 // This is also a null log which does not output anything.
 class ARROW_EXPORT ArrowLogBase {
  public:
-  virtual ~ArrowLogBase() {}
+  virtual ~ArrowLogBase() noexcept(false) {}
 
   virtual bool IsEnabled() const { return false; }
 
@@ -168,7 +168,7 @@ class ARROW_EXPORT ArrowLogBase {
 class ARROW_EXPORT ArrowLog : public ArrowLogBase {
  public:
   ArrowLog(const char* file_name, int line_number, ArrowLogLevel severity);
-  ~ArrowLog() override;
+  ~ArrowLog() noexcept(false) override;
 
   /// Return whether or not current logging instance is enabled.
   ///

cpp/src/arrow/util/mutex.cc

@@ -19,7 +19,7 @@
 
 #include <mutex>
 
-#ifndef _WIN32
+#if !defined( _WIN32) && !defined(__ppc64__)
 #  include <pthread.h>
 #  include <atomic>
 #endif
@@ -59,7 +59,7 @@ Mutex::Guard Mutex::Lock() {
 
 Mutex::Mutex() : impl_(new Impl, [](Impl* impl) { delete impl; }) {}
 
-#ifndef _WIN32
+#if !defined( _WIN32) && !defined(__ppc64__)
 namespace {
 
 struct AfterForkState {
@@ -71,19 +71,42 @@ struct AfterForkState {
   // The leak (only in child processes) is a small price to pay for robustness.
   Mutex* mutex = nullptr;
 
+  enum State {
+    INITIALIZED,
+    IN_PROCESS,
+    NOT_INITIALIZED,
+  };
+
+  std::atomic_int state = INITIALIZED;
+
  private:
   AfterForkState() {
     pthread_atfork(/*prepare=*/nullptr, /*parent=*/nullptr, /*child=*/&AfterFork);
   }
 
-  static void AfterFork() { instance.mutex = new Mutex; }
+  static void AfterFork() { instance.state.store(NOT_INITIALIZED); }
+
 };
 
 AfterForkState AfterForkState::instance;
 }  // namespace
 
-Mutex* GlobalForkSafeMutex() { return AfterForkState::instance.mutex; }
-#endif  // _WIN32
+Mutex* GlobalForkSafeMutex() {
+  if (AfterForkState::instance.state.load() == AfterForkState::State::INITIALIZED) {
+    return AfterForkState::instance.mutex;
+  }
+
+  int expected = AfterForkState::State::NOT_INITIALIZED;
+  if (AfterForkState::instance.state.compare_exchange_strong(expected, AfterForkState::State::IN_PROCESS)) {
+    AfterForkState::instance.mutex = new Mutex;
+    AfterForkState::instance.state.store(AfterForkState::State::INITIALIZED);
+  } else {
+    while (AfterForkState::instance.state.load() != AfterForkState::State::INITIALIZED);
+  }
+
+  return AfterForkState::instance.mutex;
+}
+#endif  // _WIN32 and __ppc64__
 
 }  // namespace util
 }  // namespace arrow

cpp/src/arrow/util/mutex.h

@@ -60,7 +60,7 @@ class ARROW_EXPORT Mutex {
   std::unique_ptr<Impl, void (*)(Impl*)> impl_;
 };
 
-#ifndef _WIN32
+#if !defined(_WIN32) && !defined(__ppc64__)
 /// Return a pointer to a process-wide, process-specific Mutex that can be used
 /// at any point in a child process.  NULL is returned when called in the parent.
 ///