Predicate Pushdown to SeqScan

Integrate the Filter operator above the SeqScan operator into SeqScan itself, allowing the system to lock only rows that satisfy the predicate.

Modify the Next() function in SeqScanExecutor:

auto SeqScanExecutor::Next(Tuple *result_tuple, RID *result_rid) -> bool {
  do {
    if (table_iterator_ == table_info_->table_->End()) {
      // SeqScan complete; for READ_COMMITTED, release all row S-locks and IS table locks early
      if (exec_ctx_->GetTransaction()->GetIsolationLevel() == IsolationLevel::READ_COMMITTED) {
        try {
          if (!exec_ctx_->GetTransaction()->GetSharedRowLockSet()->empty()) {
            const auto &locked_rows = exec_ctx_->GetTransaction()->GetSharedRowLockSet()->at(table_info_->oid_);
            for (auto row_rid : locked_rows) {
              exec_ctx_->GetLockManager()->UnlockRow(exec_ctx_->GetTransaction(), table_info_->oid_, row_rid);
            }
          }
          exec_ctx_->GetLockManager()->UnlockTable(exec_ctx_->GetTransaction(), table_info_->oid_);
        } catch (TransactionAbortException &e) {
          throw ExecutionException("SeqScan Executor Unlock Row/Table Lock Failed" + e.GetInfo());
        }
      }
      return false;
    }
    *result_tuple = *table_iterator_++;
    *result_rid = result_tuple->GetRid();
  } while (plan_->filter_predicate_ != nullptr && 
           !plan_->filter_predicate_->Evaluate(result_tuple, table_info_->schema_).GetAs<bool>());

  // For isolation levels other than READ_UNCOMMITTED, acquire row S-lock
  if (exec_ctx_->GetTransaction()->GetIsolationLevel() != IsolationLevel::READ_UNCOMMITTED) {
    try {
      if (!exec_ctx_->GetLockManager()->LockRow(exec_ctx_->GetTransaction(), LockManager::LockMode::SHARED, 
                                                table_info_->oid_, *result_rid)) {
        throw ExecutionException("SeqScan Executor Get S Row Lock Failed");
      }
    } catch (TransactionAbortException &e) {
      throw ExecutionException("SeqScan Executor Get S Row Lock Failed" + e.GetInfo());
    }
  }
  return true;
}

Enable the OptimizeMergeFilterScan rule in the Optimizer:

  auto optimized = plan;
  optimized = OptimizeMergeProjection(optimized);
  optimized = OptimizeMergeFilterScan(optimized);
  optimized = OptimizeMergeFilterNLJ(optimized);
  optimized = OptimizeNLJAsIndexJoin(optimized);
  optimized = OptimizeNLJAsHashJoin(optimized);
  optimized = OptimizeOrderByAsIndexScan(optimized);
  optimized = OptimizeSortLimitAsTopN(optimized);
  return optimized;
}

Implementing UpdateExecutor

Implement the Update operator to modify tuples in-place without requiring Delete + Insert operations. This is straightforward: acquire IX locks on the table and X locks on rows. Return the count of updated rows. Locks are released during Commit/Abort automatically.

src/include/execution/executors/update_executor.h member variables:

/** Update plan node to be executed */
const UpdatePlanNode *plan_;
/** Metadata for the table to update */
const TableInfo *table_info_;
/** Child executor to retrieve values from */
std::unique_ptr<AbstractExecutor> child_executor_;
bool finished_ = false;

src/execution/update_executor.cpp:

UpdateExecutor::UpdateExecutor(ExecutorContext *exec_ctx, const UpdatePlanNode *plan, 
                                std::unique_ptr<AbstractExecutor> &&child)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child)) {
  // As of Fall 2022, update executor is not required for perfect score in project 3 / project 4.
}

void UpdateExecutor::Init() {
  child_executor_->Init();
  table_info_ = exec_ctx_->GetCatalog()->GetTable(plan_->TableOid());
  // Acquire IX lock on table (should be upgraded)
  try {
    if (!exec_ctx_->GetLockManager()->LockTable(exec_ctx_->GetTransaction(), 
                                                LockManager::LockMode::INTENTION_EXCLUSIVE, 
                                                table_info_->oid_)) {
      throw ExecutionException("Update Executor Get IX Table Lock Failed");
    }
  } catch (TransactionAbortException &e) {
    throw ExecutionException("Update Executor Get IX Table Lock Failed");
  }
}

auto UpdateExecutor::Next(Tuple *result_tuple, RID *result_rid) -> bool {
  if (finished_) {
    return false;
  }
  Tuple source_tuple;
  Transaction *txn = exec_ctx_->GetTransaction();
  int32_t update_count = 0;

  while (child_executor_->Next(&source_tuple, result_rid)) {
    // Acquire X lock on row (should be upgraded)
    try {
      if (!exec_ctx_->GetLockManager()->LockRow(txn, LockManager::LockMode::EXCLUSIVE, 
                                                table_info_->oid_, *result_rid)) {
        throw ExecutionException("Update Executor Get X Row Lock Failed");
      }
    } catch (TransactionAbortException &e) {
      throw ExecutionException("Update Executor Get X Row Lock Failed" + e.GetInfo());
    }

    std::vector<Value> updated_values{};
    updated_values.reserve(child_executor_->GetOutputSchema().GetColumnCount());
    for (const auto &expr : plan_->target_expressions_) {
      updated_values.emplace_back(expr->Evaluate(&source_tuple, child_executor_->GetOutputSchema()));
    }
    Tuple new_tuple = Tuple{updated_values, &child_executor_->GetOutputSchema()};
    if (!table_info_->table_->UpdateTuple(new_tuple, *result_rid, txn)) {
      throw std::logic_error("The tuple could not be found.");
    }
    update_count++;
  }
  *result_tuple = Tuple{std::vector<Value>{Value(TypeId::INTEGER, update_count)}, &GetOutputSchema()};
  finished_ = true;
  return true;
}

Finally, enable the corresponding configuration in tools/terrier_bench/terrier_bench_config.h to instruct Terriers to use UPDATE for exchanging NFTs:

#define TERRIER_BENCH_ENABLE_UPDATE
// #define TERRIER_BENCH_ENABLE_INDEX

Using Index Scans

For queries like the following with an index on the id column, the system can leverage the index to avoid full table scans. This optimization provides the most significant performance improvement, potentially increasing QPS from dozens to tens of thousands.

SELECT ... WHERE id = 1;

When implementing this optimization, only the specific query patterns used in tests need to be handled. The primary query to optimize is:

SELECT * FROM nft WHERE id = <nft_id>

This involves converting the underlying SeqScanExecutor to IndexScanExecutor through an optimization rule. The IndexScanExecutor must implement point query logic. The optimization rule should match patterns like:

Filter
  |
SeqScan

The Filter predicate should be in the form x=a, where x is a ColumnValue with an index, and a is a ConstantValue. This optimization rule must execute before MergeFilterScan, otherwise the Filter would be merged directly into SeqScan. (This implementation places the rule before MergeFilterScan).

After successful matching, extract a in IndexScanExecutor and use ScanKey(tuple{a}) for querying. For point queries, acquire IS table locks and S row locks (unless READ_UNCOMMITTED). For READ_COMMITTED, release locks early when no more data is available.

First, add the optimization rule:

  std::vector<AbstractPlanNodeRef> children;
  for (const auto &child : plan->GetChildren()) {
    children.emplace_back(OptimizeMergeFilterIndexScan(child));
  }
  auto optimized_plan = plan->CloneWithChildren(std::move(children));

  if (optimized_plan->GetType() == PlanType::Filter) {
    const auto &filter_plan = dynamic_cast<const FilterPlanNode &>(*optimized_plan);
    BUSTUB_ASSERT(optimized_plan->children_.size() == 1, "must have exactly one children");
    const auto &child_plan = *optimized_plan->children_[0];

    if (child_plan.GetType() == PlanType::SeqScan) {
      const auto &seq_scan_plan = dynamic_cast<const SeqScanPlanNode &>(child_plan);
      const auto *table_info = catalog_.GetTable(seq_scan_plan.GetTableOid());
      const auto indexes = catalog_.GetTableIndexes(table_info->name_);

      if (const auto *expr = dynamic_cast<const ComparisonExpression *>(filter_plan.GetPredicate().get());
          expr != nullptr) {
        if (expr->comp_type_ == ComparisonType::Equal) {
          if (const auto *left_expr = dynamic_cast<const ColumnValueExpression *>(expr->children_[0].get());
              left_expr != nullptr) {
            if (const auto *right_expr = dynamic_cast<const ConstantValueExpression *>(expr->children_[1].get());
                right_expr != nullptr) {
              for (const auto index : indexes) {
                const auto &columns = index->key_schema_.GetColumns();
                if (columns.size() == 1 &&
                    columns[0].GetName() == table_info->schema_.GetColumn(left_expr->GetColIdx()).GetName()) {
                  return std::make_shared<IndexScanPlanNode>(optimized_plan->output_schema_, 
                                                             index->index_oid_,
                                                             filter_plan.GetPredicate());
                }
              }
            }
          }
        }
      }
    }
  }
  return optimized_plan;
}

Apply this rule in OptimizeCustom():

  auto optimized = plan;
  optimized = OptimizeMergeProjection(optimized);
  optimized = OptimizeMergeFilterIndexScan(optimized);  // Push predicate to index if available (p4 Leaderboard Task)
  optimized = OptimizeMergeFilterScan(optimized);
  optimized = OptimizeMergeFilterNLJ(optimized);
  optimized = OptimizeNLJAsIndexJoin(optimized);
  optimized = OptimizeNLJAsHashJoin(optimized);
  optimized = OptimizeOrderByAsIndexScan(optimized);
  optimized = OptimizeSortLimitAsTopN(optimized);
  return optimized;
}

Next, modify IndexScanExecutor to execute the optimized plan.

First, add a member variable to IndexScanPlanNode (similar to SeqScanPlanNode) to store the predicate for point queries:

AbstractExpressionRef filter_predicate_;

Add member variables to IndexScanExecutor.h to store RIDs from point queries:

std::vector<RID> matching_rids_;
std::vector<RID>::iterator rid_iterator_;

In Init(), check if a predicate exists. If so, use the index's ScanKey to find matching keys and store all RIDs in matching_rids_. The Next() function then iterates through matching_rids_ to return results:

IndexScanExecutor::IndexScanExecutor(ExecutorContext *exec_ctx, const IndexScanPlanNode *plan)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      index_info_(exec_ctx_->GetCatalog()->GetIndex(plan_->GetIndexOid())),
      table_info_(exec_ctx_->GetCatalog()->GetTable(index_info_->table_name_)),
      index_(dynamic_cast<BPlusTreeIndexForOneIntegerColumn *>(index_info_->index_.get())),
      index_iterator_(plan_->filter_predicate_ != nullptr ? BPlusTreeIndexIteratorForOneIntegerColumn()
                                                          : index_->GetBeginIterator()) {}

void IndexScanExecutor::Init() {
  // Acquire IS table lock for non-READ_UNCOMMITTED isolation levels
  if (exec_ctx_->GetTransaction()->GetIsolationLevel() != IsolationLevel::READ_UNCOMMITTED) {
    try {
      if (!exec_ctx_->GetLockManager()->LockTable(exec_ctx_->GetTransaction(), 
                                                  LockManager::LockMode::INTENTION_SHARED, 
                                                  table_info_->oid_)) {
        throw ExecutionException("IndexScan Executor Get IS Table Lock Failed");
      }
    } catch (TransactionAbortException &e) {
      throw ExecutionException("IndexScan Executor Get IS Table Lock Failed" + e.GetInfo());
    }
  }

  if (plan_->filter_predicate_ != nullptr) {
    // Extract predicate and perform index lookup
    const auto *constant_expr = dynamic_cast<const ConstantValueExpression *>(plan_->filter_predicate_->children_[1].get());
    Value search_value = constant_expr->val_;
    index_->ScanKey(Tuple{{search_value}, index_info_->index_->GetKeySchema()}, 
                    &matching_rids_, exec_ctx_->GetTransaction());
    rid_iterator_ = matching_rids_.begin();
  }
}

auto IndexScanExecutor::Next(Tuple *result_tuple, RID *result_rid) -> bool {
  // Point query with predicate pushdown
  if (plan_->filter_predicate_ != nullptr) {
    if (rid_iterator_ == matching_rids_.end()) {
      // IndexScan complete; for READ_COMMITTED, release locks early
      if (exec_ctx_->GetTransaction()->GetIsolationLevel() == IsolationLevel::READ_COMMITTED) {
        try {
          if (!exec_ctx_->GetTransaction()->GetSharedRowLockSet()->empty()) {
            const auto &locked_rows = exec_ctx_->GetTransaction()->GetSharedRowLockSet()->at(table_info_->oid_);
            for (auto row_rid : locked_rows) {
              exec_ctx_->GetLockManager()->UnlockRow(exec_ctx_->GetTransaction(), table_info_->oid_, row_rid);
            }
          }
          exec_ctx_->GetLockManager()->UnlockTable(exec_ctx_->GetTransaction(), table_info_->oid_);
        } catch (TransactionAbortException &e) {
          throw ExecutionException("IndexScan Executor Unlock Row/Table Lock Failed" + e.GetInfo());
        }
      }
      return false;
    }
    *result_rid = *rid_iterator_;
    ++rid_iterator_;

    // Acquire row S-lock for non-READ_UNCOMMITTED isolation levels
    if (exec_ctx_->GetTransaction()->GetIsolationLevel() != IsolationLevel::READ_UNCOMMITTED) {
      try {
        if (!exec_ctx_->GetLockManager()->LockRow(exec_ctx_->GetTransaction(), LockManager::LockMode::SHARED, 
                                                  table_info_->oid_, *result_rid)) {
          throw ExecutionException("IndexScan Executor Get S Row Lock Failed");
        }
      } catch (TransactionAbortException &e) {
        throw ExecutionException("IndexScan Executor Get S Row Lock Failed" + e.GetInfo());
      }
    }
    return table_info_->table_->GetTuple(*result_rid, result_tuple, exec_ctx_->GetTransaction());
  }

  // Full index scan without predicate
  if (index_iterator_ == index_->GetEndIterator()) {
    // Release locks for READ_COMMITTED
    if (exec_ctx_->GetTransaction()->GetIsolationLevel() == IsolationLevel::READ_COMMITTED) {
      try {
        if (!exec_ctx_->GetTransaction()->GetSharedRowLockSet()->empty()) {
          const auto &locked_rows = exec_ctx_->GetTransaction()->GetSharedRowLockSet()->at(table_info_->oid_);
          for (auto row_rid : locked_rows) {
            exec_ctx_->GetLockManager()->UnlockRow(exec_ctx_->GetTransaction(), table_info_->oid_, row_rid);
          }
        }
        exec_ctx_->GetLockManager()->UnlockTable(exec_ctx_->GetTransaction(), table_info_->oid_);
      } catch (TransactionAbortException &e) {
        throw ExecutionException("IndexScan Executor Unlock Row/Table Lock Failed" + e.GetInfo());
      }
    }
    return false;
  }
  *result_rid = (*index_iterator_).second;
  ++index_iterator_;

  // Acquire row S-lock for non-READ_UNCOMMITTED isolation levels
  if (exec_ctx_->GetTransaction()->GetIsolationLevel() != IsolationLevel::READ_UNCOMMITTED) {
    try {
      if (!exec_ctx_->GetLockManager()->LockRow(exec_ctx_->GetTransaction(), LockManager::LockMode::SHARED, 
                                                table_info_->oid_, *result_rid)) {
        throw ExecutionException("IndexScan Executor Get S Row Lock Failed");
      }
    } catch (TransactionAbortException &e) {
      throw ExecutionException("IndexScan Executor Get S Row Lock Failed" + e.GetInfo());
    }
  }
  return table_info_->table_->GetTuple(*result_rid, result_tuple, exec_ctx_->GetTransaction());
}

Finally, enable the configuration in tools/terrier_bench/terrier_bench_config.h to instruct Terriers to create indexes before exchanging NFTs:

#define TERRIER_BENCH_ENABLE_UPDATE
#define TERRIER_BENCH_ENABLE_INDEX

Optimizing GrantLock()

For additional performance gains, optimize the GrantLock() function:

Original logic: Both LockTable() and LockRow() append new requests to the end of the queue (the difference being that lock upgrades delete old requests from the queue). Then GrantLock() iterates through the queue to collect granted locks and waiting locks.

The reason for iterating through the queue is that upgrading requests have highest priority and are appended to the end, so granted requests are not necessarily at the front of the queue.

if (!granted_lock_set.empty()) {
    if (current lock_request compatible with granted_lock_set) {
        // If there's an upgrading request, it has highest priority (unless it's the current request)
        if (lock_request_queue->upgrading_ != INVALID_TXN_ID) {
            return lock_request_queue->upgrading_ == lock_request->txn_id_;
        }
        // No upgrading request; grant if noone is waiting, otherwise waiting locks have priority
        return wait_lock_set.empty();
    }
    return false;
}
return current lock_request compatible with wait_lock_set

Optimized approach:

In LockTable() and LockRow(), insert upgrading requests at the first waiting position instead of appending to the end. Regular requests still go to the end.

if (lock_request_queue->upgrading_ == txn->GetTransactionId()) {
  auto lr_iter = lock_request_queue->request_queue_.begin();
  while (lr_iter != lock_request_queue->request_queue_.end() && (*lr_iter)->granted_) {
    ++lr_iter;
  }
  lock_request_queue->request_queue_.insert(lr_iter, lock_request);
} else {
  lock_request_queue->request_queue_.push_back(lock_request);
}

This eliminates the need to iterate through the entire queue in GrantLock(). Since granted locks are always at the front of the queue, we can process them sequentially. When encountering a non-granted request, it must be the first waiting request to be granted. For upgrading requests, they are guaranteed to be first in the waiting queue.

The logic "only the first waiting request gets granted" doesn't break compatibility: when the first waiter acquires a non-X lock, it calls notify_all, allowing subsequent waiters to be evaluated against granted locks.

                             const std::shared_ptr<LockRequestQueue> &lock_request_queue) -> bool {
  for (auto &lr : lock_request_queue->request_queue_) {
    if (lr->granted_) {
      switch (lock_request->lock_mode_) {
        case LockMode::SHARED:
          if (lr->lock_mode_ == LockMode::INTENTION_EXCLUSIVE ||
              lr->lock_mode_ == LockMode::SHARED_INTENTION_EXCLUSIVE ||
              lr->lock_mode_ == LockMode::EXCLUSIVE) {
            return false;
          }
          break;
        case LockMode::EXCLUSIVE:
          return false;
        case LockMode::INTENTION_SHARED:
          if (lr->lock_mode_ == LockMode::EXCLUSIVE) {
            return false;
          }
          break;
        case LockMode::INTENTION_EXCLUSIVE:
          if (lr->lock_mode_ == LockMode::SHARED ||
              lr->lock_mode_ == LockMode::SHARED_INTENTION_EXCLUSIVE ||
              lr->lock_mode_ == LockMode::EXCLUSIVE) {
            return false;
          }
          break;
        case LockMode::SHARED_INTENTION_EXCLUSIVE:
          if (lr->lock_mode_ != LockMode::INTENTION_SHARED) {
            return false;
          }
          break;
      }
    } else if (lock_request.get() != lr.get()) {
      return false;
    } else {
      return true;
    }
  }
  return false;
}

Final Results

After submitting to GradeScope, the ranking improved from 7th to 5th place.