fix: resolve standalone join failures and catalog error handling

Author: poyrazK

src/catalog/catalog.cpp

@@ -145,10 +145,7 @@ oid_t Catalog::create_table(const std::string& table_name, std::vector<ColumnInf
 oid_t Catalog::create_table_local(const std::string& table_name, std::vector<ColumnInfo> columns,
                                   std::vector<ShardInfo> shards) {
     if (table_exists_by_name(table_name)) {
-        std::cerr << "--- [Catalog] create_table_local: Table already exists " << table_name
-                  << " ---" << std::endl;
-        auto meta_opt = get_table_by_name(table_name);
-        return (*meta_opt)->table_id;
+        throw std::runtime_error("Table already exists: " + table_name);
     }
 
     auto table = std::make_unique<TableInfo>();

src/executor/operator.cpp

@@ -108,8 +108,12 @@ BufferScanOperator::BufferScanOperator(std::string context_id, std::string table
     : Operator(OperatorType::BufferScan),
       context_id_(std::move(context_id)),
       table_name_(std::move(table_name)),
-      data_(std::move(data)),
-      schema_(std::move(schema)) {}
+      data_(std::move(data)) {
+    /* Qualify columns in buffer schema */
+    for (const auto& col : schema.columns()) {
+        schema_.add_column(table_name_ + "." + col.name(), col.type(), col.nullable());
+    }
+}
 
 bool BufferScanOperator::next(Tuple& out_tuple) {
     if (current_index_ >= data_.size()) {
@@ -675,7 +679,6 @@ bool HashJoinOperator::next(Tuple& out_tuple) {
             left_tuple_ = std::move(next_left);
             left_had_match_ = false;
             const common::Value key = left_key_->evaluate(&(left_tuple_.value()), &left_schema);
-            std::cerr << "--- [HashJoin] Probing: key=" << key.to_string() << " ---" << std::endl;
 
             /* Look up in hash table */
             auto range = hash_table_.equal_range(key.to_string());

src/executor/query_executor.cpp

@@ -670,7 +670,7 @@ QueryResult QueryExecutor::execute_update(const parser::UpdateStatement& stmt,
             if (log_manager_ != nullptr && txn != nullptr) {
                 recovery::LogRecord log(txn->get_id(), txn->get_prev_lsn(),
                                         recovery::LogRecordType::MARK_DELETE, table_name, op.rid,
-                                        op.old_tuple);
+                                        old_tuple);
                 const auto lsn = log_manager_->append_log_record(log);
                 txn->set_prev_lsn(lsn);
             }
@@ -724,12 +724,12 @@ std::unique_ptr<Operator> QueryExecutor::build_plan(const parser::SelectStatemen
     if (cluster_manager_ != nullptr &&
         cluster_manager_->has_shuffle_data(context_id_, base_table_name)) {
         auto data = cluster_manager_->fetch_shuffle_data(context_id_, base_table_name);
-        /* We need a schema for the buffered data. */
+        /* We need a schema for the buffered data. Use unqualified names as BufferScan will qualify them. */
         auto meta_opt = catalog_.get_table_by_name(base_table_name);
         Schema buffer_schema;
         if (meta_opt.has_value()) {
             for (const auto& col : meta_opt.value()->columns) {
-                buffer_schema.add_column(base_table_name + "." + col.name, col.type);
+                buffer_schema.add_column(col.name, col.type);
             }
         }
         std::cerr << "--- [BuildPlan] Table " << base_table_name
@@ -746,7 +746,7 @@ std::unique_ptr<Operator> QueryExecutor::build_plan(const parser::SelectStatemen
 
         Schema base_schema;
         for (const auto& col : base_table_meta->columns) {
-            base_schema.add_column(base_table_name + "." + col.name, col.type);
+            base_schema.add_column(col.name, col.type);
         }
 
         /* Index Selection Optimization:
@@ -825,7 +825,7 @@ std::unique_ptr<Operator> QueryExecutor::build_plan(const parser::SelectStatemen
             Schema buffer_schema;
             if (meta_opt.has_value()) {
                 for (const auto& col : meta_opt.value()->columns) {
-                    buffer_schema.add_column(join_table_name + "." + col.name, col.type);
+                    buffer_schema.add_column(col.name, col.type);
                 }
             }
             std::cerr << "--- [BuildPlan] JOIN Table " << join_table_name
@@ -842,7 +842,7 @@ std::unique_ptr<Operator> QueryExecutor::build_plan(const parser::SelectStatemen
 
             Schema join_schema;
             for (const auto& col : join_table_meta->columns) {
-                join_schema.add_column(join_table_name + "." + col.name, col.type);
+                join_schema.add_column(col.name, col.type);
             }
 
             join_scan = std::make_unique<SeqScanOperator>(