Fix error print (#2042)

### What problem does this PR solve?

Print critical log when raising unrecoverable error.

### Type of change

- [x] Refactoring

---------

Signed-off-by: Jin Hai <[email protected]>

.github/workflows/slow_test.yml

@@ -25,7 +25,7 @@ concurrency:
 jobs:
   slow_tests:
     name: run slow test
-    if: ${{ github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'invalid') }}
+    if: ${{ github.event_name != 'pull_request' || contains(github.event.pull_request.labels.*.name, 'slow-test') }}
     runs-on: ["self-hosted", "slow-test" ]
     steps:
 

src/common/utility/exception.cpp

@@ -48,7 +48,9 @@ void PrintStacktrace(const String &err_msg) {
 
 void RecoverableError(Status status, const char *file_name, u32 line) {
     status.AppendMessage(fmt::format("@{}:{}", infinity::TrimPath(file_name), line));
-    LOG_ERROR(status.message());
+    if (IS_LOGGER_INITIALIZED()) {
+        LOG_ERROR(status.message());
+    }
     throw RecoverableException(status);
 }
 
@@ -64,7 +66,9 @@ void UnrecoverableError(const String &message, const char *file_name, u32 line)
         String error_msg = cleanup_tracer->GetCleanupInfo();
         LOG_ERROR(std::move(error_msg));
     }
-
+    if (IS_LOGGER_INITIALIZED()) {
+        LOG_CRITICAL(message);
+    }
     Logger::Flush();
     PrintStacktrace(message);
     throw UnrecoverableException(fmt::format("{}@{}:{}", message, infinity::TrimPath(file_name), line));
@@ -73,12 +77,17 @@ void UnrecoverableError(const String &message, const char *file_name, u32 line)
 #else
 
 void RecoverableError(Status status) {
-    LOG_ERROR(status.message());
+    if (IS_LOGGER_INITIALIZED()) {
+        LOG_ERROR(status.message());
+    }
     throw RecoverableException(status);
 }
 
 void UnrecoverableError(const String &message) {
-    LOG_CRITICAL(message);
+    if (IS_LOGGER_INITIALIZED()) {
+        LOG_CRITICAL(message);
+    }
+    Logger::Flush();
     throw UnrecoverableException(message);
 }
 

src/executor/operator/physical_export.cpp

@@ -555,7 +555,6 @@ SizeT PhysicalExport::ExportToPARQUET(QueryContext *query_context, ExportOperato
                         column_vectors.emplace_back(block_entry->GetColumnBlockEntry(select_column_idx)->GetConstColumnVector(buffer_manager));
                         if (column_vectors[block_column_idx].Size() != block_row_count) {
                             String error_message = "Unmatched row_count between block and block_column";
-                            LOG_CRITICAL(error_message);
                             UnrecoverableError(error_message);
                         }
                     }
@@ -574,7 +573,6 @@ SizeT PhysicalExport::ExportToPARQUET(QueryContext *query_context, ExportOperato
             auto status = file_writer->WriteRecordBatch(*block_batch);
             if (!status.ok()) {
                 String error_message = fmt::format("Failed to write record batch to parquet file: {}", status.message());
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
             row_count += block_row_count;
@@ -584,7 +582,6 @@ SizeT PhysicalExport::ExportToPARQUET(QueryContext *query_context, ExportOperato
     auto status = file_writer->Close();
     if (!status.ok()) {
         String error_message = fmt::format("Failed to close parquet file: {}", status.message());
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
     LOG_DEBUG(fmt::format("Export to PARQUET, db {}, table {}, file: {}, row: {}", schema_name_, table_name_, file_path_, row_count));
@@ -784,7 +781,6 @@ SharedPtr<arrow::DataType> PhysicalExport::GetArrowType(ColumnDef *column_def) {
         case LogicalType::kEmptyArray:
         case LogicalType::kInvalid: {
             String error_message = "Invalid data type";
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
     }
@@ -988,7 +984,6 @@ SharedPtr<arrow::Array> PhysicalExport::BuildArrowArray(ColumnDef *column_def, c
                 }
                 case EmbeddingDataType::kElemInvalid: {
                     String error_message = "Invalid embedding data type: EmbeddingDataType::kElemInvalid";
-                    LOG_CRITICAL(error_message);
                     UnrecoverableError(error_message);
                     break;
                 }
@@ -1031,7 +1026,6 @@ SharedPtr<arrow::Array> PhysicalExport::BuildArrowArray(ColumnDef *column_def, c
         case LogicalType::kEmptyArray:
         case LogicalType::kInvalid: {
             String error_message = "Invalid data type";
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
     }
@@ -1045,7 +1039,6 @@ SharedPtr<arrow::Array> PhysicalExport::BuildArrowArray(ColumnDef *column_def, c
     auto status = array_builder->Finish(&array);
     if (!status.ok()) {
         String error_message = fmt::format("Failed to build arrow array: {}", status.message());
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
     return array;

src/executor/operator/physical_import.cpp

@@ -773,7 +773,7 @@ void PhysicalImport::CSVRowHandler(void *context) {
     // if column count is larger than columns defined from schema, extra columns are abandoned
     if (column_count > table_entry->ColumnCount()) {
         UniquePtr<String> err_msg = MakeUnique<String>(
-            fmt::format("CSV file column count isn't match with table schema, row id: {}, column_count = {}, table_entry->ColumnCount = {}.",
+            fmt::format("CSV file column count isn't match with table schema, row id: {}, column_count: {}, table_entry->ColumnCount: {}.",
                         parser_context->row_count_,
                         column_count,
                         table_entry->ColumnCount()));
@@ -800,7 +800,7 @@ void PhysicalImport::CSVRowHandler(void *context) {
                 auto &column_vector = parser_context->column_vectors_[column_idx];
                 column_vector.AppendByConstantExpr(const_expr);
             } else {
-                Status status = Status::ImportFileFormatError(fmt::format("Column {} is empty.", column_def->name_));
+                Status status = Status::ImportFileFormatError(fmt::format("No value in column {} in CSV of row number: {}", column_def->name_, parser_context->row_count_));
                 RecoverableError(status);
             }
         }
@@ -812,7 +812,7 @@ void PhysicalImport::CSVRowHandler(void *context) {
             auto const_expr = dynamic_cast<ConstantExpr *>(column_def->default_expr_.get());
             column_vector.AppendByConstantExpr(const_expr);
         } else {
-            Status status = Status::ImportFileFormatError(fmt::format("Column {} is empty.", column_def->name_));
+            Status status = Status::ImportFileFormatError(fmt::format("No value in column {} in CSV of row number: {}", column_def->name_, parser_context->row_count_));
             RecoverableError(status);
         }
     }
@@ -1977,7 +1977,6 @@ void PhysicalImport::ParquetValueHandler(const SharedPtr<arrow::Array> &array, C
         case LogicalType::kEmptyArray:
         case LogicalType::kInvalid: {
             String error_message = "Not implement: Invalid data type.";
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
     }

src/main/logger.cppm

@@ -44,7 +44,7 @@ public:
     Shutdown();
 };
 
-inline bool IS_LOGGER_INITIALIZED() { return infinity_logger.get() != nullptr; }
+export inline bool IS_LOGGER_INITIALIZED() { return infinity_logger.get() != nullptr; }
 
 export inline bool SHOULD_LOG_TRACE() { return IS_LOGGER_INITIALIZED() && infinity_logger->should_log(spdlog::level::level_enum::trace); }
 

src/network/peer_server_thrift_service.cpp

@@ -47,7 +47,6 @@ void PeerServerThriftService::Register(infinity_peer_server::RegisterResponse &r
             }
             default: {
                 String error_message = fmt::format("Invalid node type: {}", infinity_peer_server::to_string(request.node_type));
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }
@@ -115,7 +114,6 @@ void PeerServerThriftService::HeartBeat(infinity_peer_server::HeartBeatResponse
             }
             default: {
                 String error_message = "Invalid node type";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }

src/network/peer_thrift_client.cpp

@@ -189,7 +189,6 @@ void PeerClient::Register(RegisterPeerTask *peer_task) {
             }
             default: {
                 String error_message = "Register to the leader";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }
@@ -224,7 +223,6 @@ void PeerClient::Unregister(UnregisterPeerTask *peer_task) {
             }
             default: {
                 String error_message = "Unregister from the leader";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }
@@ -271,7 +269,6 @@ void PeerClient::HeartBeat(HeartBeatPeerTask *peer_task) {
             }
             default: {
                 String error_message = "Heartbeat: error in data transfer to leader";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }
@@ -311,7 +308,6 @@ void PeerClient::HeartBeat(HeartBeatPeerTask *peer_task) {
                 }
                 default: {
                     String error_message = "Invalid role type";
-                    LOG_CRITICAL(error_message);
                     UnrecoverableError(error_message);
                 }
             }
@@ -326,7 +322,6 @@ void PeerClient::HeartBeat(HeartBeatPeerTask *peer_task) {
                 }
                 default: {
                     String error_message = "Invalid node status";
-                    LOG_CRITICAL(error_message);
                     UnrecoverableError(error_message);
                 }
             }
@@ -357,7 +352,6 @@ void PeerClient::SyncLogs(SyncLogTask *peer_task) {
             }
             default: {
                 String error_message = "Synlog: error in data transfer to follower or learner";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
         }

src/storage/column_vector/value.cpp

@@ -1656,7 +1656,6 @@ void Value::AppendToArrowArray(const SharedPtr<DataType> &data_type, SharedPtr<a
             Span<char> data_span = this->GetEmbedding();
             if (data_span.size() != embedding_info->Size()) {
                 String error_message = "Embedding data size mismatch.";
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
             const EmbeddingT embedding(const_cast<char *>(data_span.data()), false);
@@ -1731,7 +1730,6 @@ void Value::AppendToArrowArray(const SharedPtr<DataType> &data_type, SharedPtr<a
         case LogicalType::kEmptyArray:
         case LogicalType::kInvalid: {
             String error_message = "Invalid data type";
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
     }

src/storage/io/local_file_handle.cpp

@@ -38,14 +38,12 @@ LocalFileHandle::~LocalFileHandle() {
 
     if(fd_ == -1) {
         String error_message = fmt::format("File was closed before or not open");
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
 
     i32 ret = close(fd_);
     if(ret == -1) {
         String error_message = fmt::format("Close file: {}, error: {}", path_, strerror(errno));
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
 
@@ -62,14 +60,12 @@ Status LocalFileHandle::Close() {
 
     if(fd_ == -1) {
         String error_message = fmt::format("File was closed before");
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
 
     i32 ret = close(fd_);
     if(ret == -1) {
         String error_message = fmt::format("Close file: {}, error: {}", path_, strerror(errno));
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
 
@@ -82,15 +78,13 @@ Status LocalFileHandle::Close() {
 Status LocalFileHandle::Append(const void *buffer, u64 nbytes) {
     if(access_mode_ != FileAccessMode::kWrite) {
         String error_message = fmt::format("File: {} isn't open.", path_);
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
     i64 written = 0;
     while (written < (i64)nbytes) {
         i64 write_count = write(fd_, (char*)buffer + written, nbytes - written);
         if (write_count == -1) {
             String error_message = fmt::format("Can't write file: {}: {}. fd: {}", path_, strerror(errno), fd_);
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
         written += write_count;
@@ -112,7 +106,6 @@ Tuple<SizeT, Status> LocalFileHandle::Read(void *buffer, u64 nbytes) {
         }
         if (read_count == -1) {
             String error_message = fmt::format("Can't read file: {}: {}", path_, strerror(errno));
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
         read_n += read_count;
@@ -130,7 +123,6 @@ Tuple<SizeT, Status> LocalFileHandle::Read(String &buffer, u64 nbytes) {
         }
         if (read_count == -1) {
             String error_message = fmt::format("Can't read file: {}: {}", path_, strerror(errno));
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
         read_n += read_count;
@@ -141,7 +133,6 @@ Tuple<SizeT, Status> LocalFileHandle::Read(String &buffer, u64 nbytes) {
 Status LocalFileHandle::Seek(u64 nbytes) {
     if ((off_t)-1 == lseek(fd_, nbytes, SEEK_SET)) {
         String error_message = fmt::format("Can't seek file: {}: {}", path_, strerror(errno));
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
     return Status::OK();

src/storage/meta/cleanup_scanner.cpp

@@ -84,7 +84,6 @@ void CleanupScanner::CleanupDir(const String &abs_dir) {
             LOG_WARN(fmt::format("Cleanup: Dir {} not found. Skip", abs_dir));
         } else {
             String error_message = fmt::format("Cleanup {} encounter unexpected error: {}", abs_dir, e.what());
-            LOG_CRITICAL(error_message);
             UnrecoverableError(error_message);
         }
     }

src/storage/meta/entry/entry_list.cppm

@@ -140,7 +140,6 @@ public:
         for (const auto &base_entry : entry_list_) {
             if (base_entry->entry_type_ != entry_type) {
                 String error_message = fmt::format("Unexpected entry type {}", ToString(entry_type));
-                LOG_CRITICAL(error_message);
                 UnrecoverableError(error_message);
             }
             if (base_entry->commit_ts_ <= max_commit_ts) {

src/storage/wal/log_file.cpp

@@ -160,7 +160,6 @@ Pair<Optional<TempWalFileInfo>, Vector<WalFileInfo>> WalFile::ParseWalFilenames(
 
     auto [entries, status] = VirtualStore::ListDirectory(wal_dir);
     if(!status.ok()) {
-        LOG_CRITICAL(status.message());
         UnrecoverableError(status.message());
     }
 

src/storage/wal/wal_manager.cpp

@@ -479,7 +479,6 @@ void WalManager::UpdateCommitState(TxnTimeStamp commit_ts, i64 wal_size) {
                                            max_commit_ts_,
                                            wal_size,
                                            wal_size_);
-        LOG_CRITICAL(error_message);
         UnrecoverableError(error_message);
     }
     max_commit_ts_ = commit_ts;