fix: load data

Author: nagisa-kunhah

src/observer/sql/executor/load_data_executor.cpp

@@ -22,6 +22,76 @@ See the Mulan PSL v2 for more details. */
 
 using namespace common;
 
+namespace {
+
+void split_line_simple(const string &line, const char terminated, vector<string> &fields)
+{
+  fields.clear();
+  string current;
+  for (char c : line) {
+    if (c == terminated) {
+      fields.emplace_back(std::move(current));
+      current.clear();
+    } else {
+      current.push_back(c);
+    }
+  }
+  fields.emplace_back(std::move(current));
+}
+
+// Split a line by `terminated`, honoring optional enclosing quotes.
+// - If `enclosed` is 0, behaves like a simple split (but preserves empty fields).
+// - If `enclosed` is non-zero, treats `enclosed ... enclosed` as a quoted field where
+//   delimiters inside quotes are not split.
+// - Inside a quoted field, `""` is treated as one `"` (CSV-style escaping).
+void split_line_with_enclosed(const string &line, const char terminated, const char enclosed, vector<string> &fields)
+{
+  if (enclosed == 0) {
+    split_line_simple(line, terminated, fields);
+    return;
+  }
+
+  fields.clear();
+  string current;
+  bool   in_quotes = false;
+
+  for (size_t i = 0; i < line.size(); i++) {
+    const char c = line[i];
+
+    if (!in_quotes && c == terminated) {
+      fields.emplace_back(std::move(current));
+      current.clear();
+      continue;
+    }
+
+    if (c == enclosed) {
+      if (in_quotes) {
+        // CSV escaping: "" -> "
+        if (i + 1 < line.size() && line[i + 1] == enclosed) {
+          current.push_back(enclosed);
+          i++;
+        } else {
+          in_quotes = false;  // consume closing quote
+        }
+      } else {
+        // Start quoted field only when quote appears at the beginning of the field.
+        if (current.empty()) {
+          in_quotes = true;
+        } else {
+          current.push_back(c);
+        }
+      }
+      continue;
+    }
+
+    current.push_back(c);
+  }
+
+  fields.emplace_back(std::move(current));
+}
+
+}  // namespace
+
 RC LoadDataExecutor::execute(SQLStageEvent *sql_event)
 {
   RC            rc         = RC::SUCCESS;
@@ -105,10 +175,10 @@ void LoadDataExecutor::load_data(
   vector<Value>  record_values(field_num);
   string         line;
   vector<string> file_values;
-  const string   delim("|");
   int            line_num        = 0;
   int            insertion_count = 0;
   RC             rc              = RC::SUCCESS;
+  const char     effective_terminated = (terminated != 0) ? terminated : '|';
   while (!fs.eof() && RC::SUCCESS == rc) {
     getline(fs, line);
     line_num++;
@@ -117,7 +187,10 @@ void LoadDataExecutor::load_data(
     }
 
     file_values.clear();
-    common::split_string(line, delim, file_values);
+    if (!line.empty() && line.back() == '\r') {
+      line.pop_back();  // handle CRLF files
+    }
+    split_line_with_enclosed(line, effective_terminated, enclosed, file_values);
     stringstream errmsg;
 
     if (table->table_meta().storage_format() == StorageFormat::ROW_FORMAT ||
@@ -141,6 +214,7 @@ void LoadDataExecutor::load_data(
   if (RC::SUCCESS == rc) {
     result_string << strrc(rc);
   }
-  LOG_INFO("load data done. row num: %s, result: %s", insertion_count, strrc(rc));
-  sql_result->set_return_code(RC::SUCCESS);
+  LOG_INFO("load data done. row num: %d, result: %s", insertion_count, strrc(rc));
+  sql_result->set_return_code(rc);
+  sql_result->set_state_string(result_string.str());
 }

src/observer/sql/operator/scalar_group_by_physical_operator.cpp

@@ -82,11 +82,19 @@ RC ScalarGroupByPhysicalOperator::open(Trx *trx)
     return rc;
   }
 
-  // 得到最终聚合后的值
-  if (group_value_) {
-    rc = evaluate(*group_value_);
+  // Even if the child operator returns no tuples (empty input), scalar aggregation
+  // should still output exactly one row (e.g. `count(*)` -> 0).
+  if (group_value_ == nullptr) {
+    AggregatorList aggregator_list;
+    create_aggregator_list(aggregator_list);
+
+    CompositeTuple composite_tuple;
+    group_value_ = make_unique<GroupValueType>(std::move(aggregator_list), std::move(composite_tuple));
   }
 
+  // 得到最终聚合后的值
+  rc = evaluate(*group_value_);
+
   emitted_ = false;
   return rc;
 }
@@ -117,4 +125,4 @@ Tuple *ScalarGroupByPhysicalOperator::current_tuple()
   }
 
   return &get<1>(*group_value_);
-}
+}

src/observer/sql/stmt/load_data_stmt.cpp

@@ -43,15 +43,34 @@ RC LoadDataStmt::create(Db *db, const LoadDataSqlNode &load_data, Stmt *&stmt)
     return RC::FILE_NOT_EXIST;
   }
 
-  if (load_data.enclosed.size() != 3) {
-    LOG_WARN("load data invalid enclosed. enclosed=%s", load_data.enclosed.c_str());
+  auto extract_char = [](const string &s, const char *name, char &out) -> RC {
+    if (s.empty()) {
+      out = 0;
+      return RC::SUCCESS;
+    }
+    if (s.size() == 1) {
+      out = s[0];
+      return RC::SUCCESS;
+    }
+    if (s.size() == 3) {
+      out = s[1];  // like "','", "'|'", "\""
+      return RC::SUCCESS;
+    }
+    LOG_WARN("load data invalid %s. %s=%s", name, name, s.c_str());
     return RC::INVALID_ARGUMENT;
+  };
+
+  char terminated = 0;
+  char enclosed   = 0;
+  rc              = extract_char(load_data.terminated, "terminated", terminated);
+  if (OB_FAIL(rc)) {
+    return rc;
   }
-  if (load_data.terminated.size() != 3) {
-    LOG_WARN("load data invalid terminated. terminated=%s", load_data.terminated.c_str());
-    return RC::INVALID_ARGUMENT;
+  rc = extract_char(load_data.enclosed, "enclosed", enclosed);
+  if (OB_FAIL(rc)) {
+    return rc;
   }
 
-  stmt = new LoadDataStmt(table, load_data.file_name.c_str(), load_data.terminated[1], load_data.enclosed[1]);
+  stmt = new LoadDataStmt(table, load_data.file_name.c_str(), terminated, enclosed);
   return rc;
 }