cmu15-445-project3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// src/include/execution/plans/abstract_plan.h
/**
 * AbstractPlanNode represents all the possible types of plan nodes in our system.
 * Plan nodes are modeled as trees, so each plan node can have a variable number of children.
 * Per the Volcano model, the plan node receives the tuples of its children.
 * The ordering of the children may matter.
 */
class AbstractPlanNode {
 public:
  /**
   * Create a new AbstractPlanNode with the specified output schema and children.
   * @param output_schema the schema for the output of this plan node
   * @param children the children of this plan node
   */
  AbstractPlanNode(const Schema *output_schema, std::vector<const AbstractPlanNode *> &&children)
      : output_schema_(output_schema), children_(std::move(children)) {}

  /** @return the type of this plan node */
  // 抽象基类
  virtual PlanType GetType() const = 0;

 private:
  /**
   * The schema for the output of this plan node. In the volcano model, every plan node will spit out tuples,
   * and this tells you what schema this plan node's tuples will have.
   */
  const Schema *output_schema_;
  /** The children of this plan node. */
  std::vector<const AbstractPlanNode *> children_;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// src/include/execution/executors/abstract_executor.h
/**
 * The AbstractExecutor implements the Volcano tuple-at-a-time iterator model.
 * This is the base class from which all executors in the BustTub execution
 * engine inherit, and defines the minimal interface that all executors support.
 */
class AbstractExecutor {
 public:
  /**
   * Construct a new AbstractExecutor instance.
   * @param exec_ctx the executor context that the executor runs with
   */
  explicit AbstractExecutor(ExecutorContext *exec_ctx) : exec_ctx_{exec_ctx} {}

  /** Virtual destructor. */
  virtual ~AbstractExecutor() = default;

  /**
   * Initialize the executor.
   * @warning This function must be called before Next() is called!
   */
  virtual void Init() = 0;

  /**
   * Yield the next tuple from this executor.
   * @param[out] tuple The next tuple produced by this executor
   * @param[out] rid The next tuple RID produced by this executor
   * @return `true` if a tuple was produced, `false` if there are no more tuples
   */
  virtual bool Next(Tuple *tuple, RID *rid) = 0;

  /** @return The schema of the tuples that this executor produces */
  virtual const Schema *GetOutputSchema() = 0;

  /** @return The executor context in which this executor runs */
  ExecutorContext *GetExecutorContext() { return exec_ctx_; }

 protected:
  /** The executor context in which the executor runs */
  ExecutorContext *exec_ctx_;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// SELECT col_a, col_b FROM test_1 WHERE col_a < 500
TEST_F(ExecutorTest, SimpleSeqScanTest) {
  // Construct query plan
  TableInfo *table_info = GetExecutorContext()->GetCatalog()->GetTable("test_1");
  const Schema &schema = table_info->schema_;
  auto *col_a = MakeColumnValueExpression(schema, 0, "colA");
  auto *col_c = MakeColumnValueExpression(schema, 0, "colC");
  auto *const500 = MakeConstantValueExpression(ValueFactory::GetIntegerValue(500));
  auto *predicate = MakeComparisonExpression(col_a, const500, ComparisonType::LessThan);
  // Create plan node
  SeqScanPlanNode plan{out_schema, predicate, table_info->oid_};

  // Execute
  std::vector<Tuple> result_set{};
  GetExecutionEngine()->Execute(&plan, &result_set, GetTxn(), GetExecutorContext());
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

  /**
   * Execute a query plan.
   * @param plan The query plan to execute
   * @param result_set The set of tuples produced by executing the plan
   * @param txn The transaction context in which the query executes
   * @param exec_ctx The executor context in which the query executes
   * @return `true` if execution of the query plan succeeds, `false` otherwise
   */
  bool Execute(const AbstractPlanNode *plan, std::vector<Tuple> *result_set, Transaction *txn,
               ExecutorContext *exec_ctx) {
    // Construct and executor for the plan
    auto executor = ExecutorFactory::CreateExecutor(exec_ctx, plan);

    // Prepare the root executor
    executor->Init();

    // Execute the query plan
    try {
      Tuple tuple;
      RID rid;
      while (executor->Next(&tuple, &rid)) {
        if (result_set != nullptr && executor->GetOutputSchema() != nullptr) {
          result_set->push_back(tuple);
        }
      }
    } catch (TransactionAbortException &e) {
      LOG_DEBUG("execute txn abort exception:%s", e.GetInfo().c_str());
      txn_mgr_->Abort(txn);
      return false;
    } catch (Exception &e) {
      LOG_DEBUG("execute exception!");
      txn_mgr_->Abort(txn);
      return false;
    }

    return true;
  }

1
2
3

    for (auto &col : output_schma->GetColumns()) {
      values.emplace_back(col.GetExpr()->Evaluate(tuple, schema));
    }

1
2
3

    if (predicate != nullptr && !predicate->Evaluate(tuple, schema).GetAs<bool>()) {
      continue;
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

auto affect_indexes = exec_ctx_->GetCatalog()->GetTableIndexes(table->name_);

// Insert/Update index
for (const auto &table_index : affect_indexes) {
    table_index->index_->InsertEntry(
        row.KeyFromTuple(schema, *table_index->index_->GetKeySchema(), table_index->index_->GetKeyAttrs()), *rid, txn);
}

// delete index
for (const auto &table_index : affect_indexes) {
    txn->GetIndexWriteSet()->emplace_back(
        IndexWriteRecord(*rid, info->oid_, WType::DELETE, *tuple, table_index->index_oid_, exec_ctx_->GetCatalog()));
    table_index->index_->DeleteEntry(
        tuple->KeyFromTuple(info->schema_, *table_index->index_->GetKeySchema(), table_index->index_->GetKeyAttrs()),
        *rid, txn);
}

1
2
3
4
5
6
7
8

// 插入
table->table_->InsertTuple(row, rid, txn)

// 更新
table->table_->UpdateTuple(row, rid, txn)
    
// 删除
table->table_->MarkDelete(rid, txn)

1
2
3
4

    if (predicate != nullptr &&
        !predicate->EvaluateJoin(&left_tuple_, left_schema, &right_tuple, right_schema).GetAs<bool>()) {
      continue;
    }

1
2
3
4
5
6

    for (auto &col : schema->GetColumns()) {
      values.emplace_back(col.GetExpr()->EvaluateJoin(&left_tuple_, left_schema, &right_tuple, right_schema));
    }

    *tuple = Tuple(values, schema);
    found = true;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

 protected:
  // The actual value item
  union Val {
    int8_t boolean_;
    int8_t tinyint_;
    int16_t smallint_;
    int32_t integer_;
    int64_t bigint_;
    double decimal_;
    uint64_t timestamp_;
    char *varlen_;
    const char *const_varlen_;
  } value_;

  union {
    uint32_t len_;
    TypeId elem_type_id_;
  } size_;

  bool manage_data_;
  // The data type
  TypeId type_id_;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

namespace bustub {

/** AggregateKey represents a key in an aggregation operation */
struct HashJoinKey {
  Value key_;
  explicit HashJoinKey(const Value &key) : key_(key) {}
  /**
   * Compares two aggregate keys for equality.
   * @param other the other aggregate key to be compared with
   * @return `true` if both aggregate keys have equivalent group-by expressions, `false` otherwise
   */
  // 这里实现的是，所有相同的key。
  // 也可以通过下面实现相同的哈希值的key，与ppt中的相同，但是后面还需进行多余的比较key是否相同。
  bool operator==(const HashJoinKey &other) const { return key_.CompareEquals(other.key_) == CmpBool::CmpTrue; }
  // bool operator==(const HashJoinKey &other) const { return bustub::HashUtil::HashValue(&key_) == bustub::HashUtil::HashValue(&other.key_); }
};

/** HashJoinValue represents a value for each tuple of same key */
struct HashJoinValue {
  /** The tuple values */
  std::vector<Tuple> values_;
};
}  // namespace bustub

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

namespace std {

/** Implements std::hash on AggregateKey */
template <>
struct hash<bustub::HashJoinKey> {
  std::size_t operator()(const bustub::HashJoinKey &hash_key) const {
    if (hash_key.key_.IsNull()) {
      return 0;
    }
    return bustub::HashUtil::HashValue(&hash_key.key_);
  }
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

/**
 * A simplified hash table that has all the necessary functionality for aggregations.
 */
class SimpleJoinHashTable {
 public:
  void Insert(const HashJoinKey &key, const Tuple &value) { ht_[key].values_.emplace_back(value); }
  HashJoinValue GetValues(const HashJoinKey &key) {
    if (ht_.count(key) != 0) {
      return ht_[key];
    }
    return HashJoinValue{};
  }

 private:
  std::unordered_map<HashJoinKey, HashJoinValue> ht_{};
};

1
2
3
4
5
6

    auto having = plan_->GetHaving();
    ++aht_iterator_;

    if (having != nullptr && !having->EvaluateAggregate(key.group_bys_, value.aggregates_).GetAs<bool>()) {
      continue;
    }