table.cpp 24.5 KB
Newer Older
羽飞's avatar
羽飞 已提交
1 2 3 4 5 6 7 8 9 10 11
/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
12
// Created by Meiyi & Wangyunlai on 2021/5/13.
羽飞's avatar
羽飞 已提交
13 14 15 16 17 18
//

#include <limits.h>
#include <string.h>
#include <algorithm>

L
Longda 已提交
19
#include "common/defs.h"
羽飞's avatar
羽飞 已提交
20 21 22 23 24 25 26 27
#include "storage/common/table.h"
#include "storage/common/table_meta.h"
#include "common/log/log.h"
#include "common/lang/string.h"
#include "storage/default/disk_buffer_pool.h"
#include "storage/common/record_manager.h"
#include "storage/common/condition_filter.h"
#include "storage/common/meta_util.h"
羽飞's avatar
羽飞 已提交
28 29
#include "storage/index/index.h"
#include "storage/index/bplus_tree_index.h"
羽飞's avatar
羽飞 已提交
30 31
#include "storage/trx/trx.h"

L
Longda 已提交
32 33
Table::Table() : data_buffer_pool_(nullptr), file_id_(-1), record_handler_(nullptr)
{}
羽飞's avatar
羽飞 已提交
34

L
Longda 已提交
35 36 37 38 39 40
Table::~Table()
{
  if (record_handler_ != nullptr) {
    delete record_handler_;
    record_handler_ = nullptr;
  }
羽飞's avatar
羽飞 已提交
41 42 43 44 45 46

  if (data_buffer_pool_ != nullptr && file_id_ >= 0) {
    data_buffer_pool_->close_file(file_id_);
    data_buffer_pool_ = nullptr;
  }

L
Longda 已提交
47 48 49 50 51 52
  for (std::vector<Index *>::iterator it = indexes_.begin(); it != indexes_.end(); ++it) {
    Index *index = *it;
    delete index;
  }
  indexes_.clear();

羽飞's avatar
羽飞 已提交
53 54 55
  LOG_INFO("Table has been closed: %s", name());
}

L
Longda 已提交
56 57 58
RC Table::create(
    const char *path, const char *name, const char *base_dir, int attribute_count, const AttrInfo attributes[])
{
羽飞's avatar
羽飞 已提交
59

L
Longda 已提交
60
  if (common::is_blank(name)) {
羽飞's avatar
羽飞 已提交
61 62 63 64 65 66
    LOG_WARN("Name cannot be empty");
    return RC::INVALID_ARGUMENT;
  }
  LOG_INFO("Begin to create table %s:%s", base_dir, name);

  if (attribute_count <= 0 || nullptr == attributes) {
L
Longda 已提交
67
    LOG_WARN("Invalid arguments. table_name=%s, attribute_count=%d, attributes=%p", name, attribute_count, attributes);
羽飞's avatar
羽飞 已提交
68 69 70 71 72 73 74 75
    return RC::INVALID_ARGUMENT;
  }

  RC rc = RC::SUCCESS;

  // 使用 table_name.table记录一个表的元数据
  // 判断表文件是否已经存在
  int fd = ::open(path, O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0600);
L
Longda 已提交
76
  if (fd < 0) {
羽飞's avatar
羽飞 已提交
77
    if (EEXIST == errno) {
L
Longda 已提交
78
      LOG_ERROR("Failed to create table file, it has been created. %s, EEXIST, %s", path, strerror(errno));
羽飞's avatar
羽飞 已提交
79 80
      return RC::SCHEMA_TABLE_EXIST;
    }
L
Longda 已提交
81
    LOG_ERROR("Create table file failed. filename=%s, errmsg=%d:%s", path, errno, strerror(errno));
羽飞's avatar
羽飞 已提交
82 83 84 85 86 87 88 89
    return RC::IOERR;
  }

  close(fd);

  // 创建文件
  if ((rc = table_meta_.init(name, attribute_count, attributes)) != RC::SUCCESS) {
    LOG_ERROR("Failed to init table meta. name:%s, ret:%d", name, rc);
L
Longda 已提交
90
    return rc;  // delete table file
羽飞's avatar
羽飞 已提交
91 92 93 94 95 96 97 98 99 100 101 102 103
  }

  std::fstream fs;
  fs.open(path, std::ios_base::out | std::ios_base::binary);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open file for write. file name=%s, errmsg=%s", path, strerror(errno));
    return RC::IOERR;
  }

  // 记录元数据到文件中
  table_meta_.serialize(fs);
  fs.close();

L
Longda 已提交
104
  std::string data_file = table_data_file(base_dir, name);
羽飞's avatar
羽飞 已提交
105 106 107 108 109 110 111 112
  data_buffer_pool_ = theGlobalDiskBufferPool();
  rc = data_buffer_pool_->create_file(data_file.c_str());
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create disk buffer pool of data file. file name=%s", data_file.c_str());
    return rc;
  }

  rc = init_record_handler(base_dir);
L
Longda 已提交
113 114 115 116 117
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create table %s due to init record handler failed.", data_file.c_str());
    // don't need to remove the data_file
    return rc;
  }
羽飞's avatar
羽飞 已提交
118 119 120 121 122 123

  base_dir_ = base_dir;
  LOG_INFO("Successfully create table %s:%s", base_dir, name);
  return rc;
}

L
Longda 已提交
124 125
RC Table::open(const char *meta_file, const char *base_dir)
{
羽飞's avatar
羽飞 已提交
126 127
  // 加载元数据文件
  std::fstream fs;
L
Longda 已提交
128
  std::string meta_file_path = std::string(base_dir) + common::FILE_PATH_SPLIT_STR + meta_file;
羽飞's avatar
羽飞 已提交
129 130
  fs.open(meta_file_path, std::ios_base::in | std::ios_base::binary);
  if (!fs.is_open()) {
L
Longda 已提交
131
    LOG_ERROR("Failed to open meta file for read. file name=%s, errmsg=%s", meta_file_path.c_str(), strerror(errno));
羽飞's avatar
羽飞 已提交
132 133 134
    return RC::IOERR;
  }
  if (table_meta_.deserialize(fs) < 0) {
L
Longda 已提交
135 136
    LOG_ERROR("Failed to deserialize table meta. file name=%s", meta_file_path.c_str());
    fs.close();
羽飞's avatar
羽飞 已提交
137 138 139 140 141 142
    return RC::GENERIC_ERROR;
  }
  fs.close();

  // 加载数据文件
  RC rc = init_record_handler(base_dir);
L
Longda 已提交
143 144 145 146 147
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to open table %s due to init record handler failed.", base_dir);
    // don't need to remove the data_file
    return rc;
  }
羽飞's avatar
羽飞 已提交
148 149 150 151 152 153 154 155

  base_dir_ = base_dir;

  const int index_num = table_meta_.index_num();
  for (int i = 0; i < index_num; i++) {
    const IndexMeta *index_meta = table_meta_.index(i);
    const FieldMeta *field_meta = table_meta_.field(index_meta->field());
    if (field_meta == nullptr) {
L
Longda 已提交
156 157 158 159 160 161
      LOG_ERROR("Found invalid index meta info which has a non-exists field. table=%s, index=%s, field=%s",
          name(),
          index_meta->name(),
          index_meta->field());
      // skip cleanup
      //  do all cleanup action in destructive Table function
羽飞's avatar
羽飞 已提交
162 163 164 165
      return RC::GENERIC_ERROR;
    }

    BplusTreeIndex *index = new BplusTreeIndex();
L
Longda 已提交
166
    std::string index_file = table_index_file(base_dir, name(), index_meta->name());
羽飞's avatar
羽飞 已提交
167 168 169 170
    rc = index->open(index_file.c_str(), *index_meta, *field_meta);
    if (rc != RC::SUCCESS) {
      delete index;
      LOG_ERROR("Failed to open index. table=%s, index=%s, file=%s, rc=%d:%s",
L
Longda 已提交
171 172 173 174 175 176 177
          name(),
          index_meta->name(),
          index_file.c_str(),
          rc,
          strrc(rc));
      // skip cleanup
      //  do all cleanup action in destructive Table function.
羽飞's avatar
羽飞 已提交
178 179 180 181 182 183 184
      return rc;
    }
    indexes_.push_back(index);
  }
  return rc;
}

L
Longda 已提交
185 186
RC Table::commit_insert(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
187 188 189
  Record record;
  RC rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
L
Longda 已提交
190
    LOG_ERROR("Failed to get record %s: %s", this->name(), rid.to_string().c_str());
羽飞's avatar
羽飞 已提交
191 192 193 194 195 196
    return rc;
  }

  return trx->commit_insert(this, record);
}

L
Longda 已提交
197 198
RC Table::rollback_insert(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
199 200 201 202

  Record record;
  RC rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
L
Longda 已提交
203
    LOG_ERROR("Failed to get record %s: %s", this->name(), rid.to_string().c_str());
羽飞's avatar
羽飞 已提交
204 205 206 207 208 209 210
    return rc;
  }

  // remove all indexes
  rc = delete_entry_of_indexes(record.data, rid, false);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to delete indexes of record(rid=%d.%d) while rollback insert, rc=%d:%s",
L
Longda 已提交
211 212 213 214 215
        rid.page_num,
        rid.slot_num,
        rc,
        strrc(rc));
    return rc;
羽飞's avatar
羽飞 已提交
216
  }
L
Longda 已提交
217 218

  rc = record_handler_->delete_record(&rid);
羽飞's avatar
羽飞 已提交
219 220 221
  return rc;
}

L
Longda 已提交
222 223
RC Table::insert_record(Trx *trx, Record *record)
{
羽飞's avatar
羽飞 已提交
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241
  RC rc = RC::SUCCESS;

  if (trx != nullptr) {
    trx->init_trx_info(this, *record);
  }
  rc = record_handler_->insert_record(record->data, table_meta_.record_size(), &record->rid);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Insert record failed. table name=%s, rc=%d:%s", table_meta_.name(), rc, strrc(rc));
    return rc;
  }

  if (trx != nullptr) {
    rc = trx->insert_record(this, record);
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to log operation(insertion) to trx");

      RC rc2 = record_handler_->delete_record(&record->rid);
      if (rc2 != RC::SUCCESS) {
L
Longda 已提交
242 243 244 245
        LOG_ERROR("Failed to rollback record data when insert index entries failed. table name=%s, rc=%d:%s",
            name(),
            rc2,
            strrc(rc2));
羽飞's avatar
羽飞 已提交
246 247 248 249 250 251 252 253 254
      }
      return rc;
    }
  }

  rc = insert_entry_of_indexes(record->data, record->rid);
  if (rc != RC::SUCCESS) {
    RC rc2 = delete_entry_of_indexes(record->data, record->rid, true);
    if (rc2 != RC::SUCCESS) {
L
Longda 已提交
255 256 257 258
      LOG_ERROR("Failed to rollback index data when insert index entries failed. table name=%s, rc=%d:%s",
          name(),
          rc2,
          strrc(rc2));
羽飞's avatar
羽飞 已提交
259 260 261 262
    }
    rc2 = record_handler_->delete_record(&record->rid);
    if (rc2 != RC::SUCCESS) {
      LOG_PANIC("Failed to rollback record data when insert index entries failed. table name=%s, rc=%d:%s",
L
Longda 已提交
263 264 265
          name(),
          rc2,
          strrc(rc2));
羽飞's avatar
羽飞 已提交
266 267 268 269 270
    }
    return rc;
  }
  return rc;
}
L
Longda 已提交
271 272 273 274
RC Table::insert_record(Trx *trx, int value_num, const Value *values)
{
  if (value_num <= 0 || nullptr == values) {
    LOG_ERROR("Invalid argument. table name: %s, value num=%d, values=%p", name(), value_num, values);
羽飞's avatar
羽飞 已提交
275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292
    return RC::INVALID_ARGUMENT;
  }

  char *record_data;
  RC rc = make_record(value_num, values, record_data);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to create a record. rc=%d:%s", rc, strrc(rc));
    return rc;
  }

  Record record;
  record.data = record_data;
  // record.valid = true;
  rc = insert_record(trx, &record);
  delete[] record_data;
  return rc;
}

L
Longda 已提交
293 294
const char *Table::name() const
{
羽飞's avatar
羽飞 已提交
295 296 297
  return table_meta_.name();
}

L
Longda 已提交
298 299
const TableMeta &Table::table_meta() const
{
羽飞's avatar
羽飞 已提交
300 301 302
  return table_meta_;
}

L
Longda 已提交
303 304
RC Table::make_record(int value_num, const Value *values, char *&record_out)
{
羽飞's avatar
羽飞 已提交
305 306
  // 检查字段类型是否一致
  if (value_num + table_meta_.sys_field_num() != table_meta_.field_num()) {
L
Longda 已提交
307
    LOG_WARN("Input values don't match the table's schema, table name:%s", table_meta_.name());
羽飞's avatar
羽飞 已提交
308 309 310 311 312 313 314 315
    return RC::SCHEMA_FIELD_MISSING;
  }

  const int normal_field_start_index = table_meta_.sys_field_num();
  for (int i = 0; i < value_num; i++) {
    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
    const Value &value = values[i];
    if (field->type() != value.type) {
L
Longda 已提交
316 317 318 319 320
      LOG_ERROR("Invalid value type. table name =%s, field name=%s, type=%d, but given=%d",
          table_meta_.name(),
          field->name(),
          field->type(),
          value.type);
羽飞's avatar
羽飞 已提交
321 322 323 324 325 326
      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
    }
  }

  // 复制所有字段的值
  int record_size = table_meta_.record_size();
L
Longda 已提交
327
  char *record = new char[record_size];
羽飞's avatar
羽飞 已提交
328 329 330 331 332 333 334 335 336 337 338

  for (int i = 0; i < value_num; i++) {
    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
    const Value &value = values[i];
    memcpy(record + field->offset(), value.data, field->len());
  }

  record_out = record;
  return RC::SUCCESS;
}

L
Longda 已提交
339 340 341
RC Table::init_record_handler(const char *base_dir)
{
  std::string data_file = table_data_file(base_dir, table_meta_.name());
羽飞's avatar
羽飞 已提交
342 343 344 345 346 347 348
  if (nullptr == data_buffer_pool_) {
    data_buffer_pool_ = theGlobalDiskBufferPool();
  }

  int data_buffer_pool_file_id;
  RC rc = data_buffer_pool_->open_file(data_file.c_str(), &data_buffer_pool_file_id);
  if (rc != RC::SUCCESS) {
L
Longda 已提交
349
    LOG_ERROR("Failed to open disk buffer pool for file:%s. rc=%d:%s", data_file.c_str(), rc, strrc(rc));
羽飞's avatar
羽飞 已提交
350 351 352 353
    return rc;
  }

  record_handler_ = new RecordFileHandler();
L
Longda 已提交
354
  rc = record_handler_->init(data_buffer_pool_, data_buffer_pool_file_id);
羽飞's avatar
羽飞 已提交
355 356
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to init record handler. rc=%d:%s", rc, strrc(rc));
L
Longda 已提交
357 358 359
    data_buffer_pool_->close_file(data_buffer_pool_file_id);
    delete record_handler_;
    record_handler_ = nullptr;
羽飞's avatar
羽飞 已提交
360 361 362 363 364 365 366 367 368 369 370 371 372
    return rc;
  }

  file_id_ = data_buffer_pool_file_id;
  return rc;
}

/**
 * 为了不把Record暴露出去,封装一下
 */
class RecordReaderScanAdapter {
public:
  explicit RecordReaderScanAdapter(void (*record_reader)(const char *data, void *context), void *context)
L
Longda 已提交
373 374
      : record_reader_(record_reader), context_(context)
  {}
羽飞's avatar
羽飞 已提交
375

L
Longda 已提交
376 377
  void consume(const Record *record)
  {
羽飞's avatar
羽飞 已提交
378 379
    record_reader_(record->data, context_);
  }
L
Longda 已提交
380

羽飞's avatar
羽飞 已提交
381 382 383 384
private:
  void (*record_reader_)(const char *, void *);
  void *context_;
};
L
Longda 已提交
385 386 387

static RC scan_record_reader_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
388 389 390 391 392
  RecordReaderScanAdapter &adapter = *(RecordReaderScanAdapter *)context;
  adapter.consume(record);
  return RC::SUCCESS;
}

L
Longda 已提交
393 394 395
RC Table::scan_record(
    Trx *trx, ConditionFilter *filter, int limit, void *context, void (*record_reader)(const char *data, void *context))
{
羽飞's avatar
羽飞 已提交
396 397 398 399
  RecordReaderScanAdapter adapter(record_reader, context);
  return scan_record(trx, filter, limit, (void *)&adapter, scan_record_reader_adapter);
}

L
Longda 已提交
400 401 402
RC Table::scan_record(
    Trx *trx, ConditionFilter *filter, int limit, void *context, RC (*record_reader)(Record *record, void *context))
{
羽飞's avatar
羽飞 已提交
403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430
  if (nullptr == record_reader) {
    return RC::INVALID_ARGUMENT;
  }

  if (0 == limit) {
    return RC::SUCCESS;
  }

  if (limit < 0) {
    limit = INT_MAX;
  }

  IndexScanner *index_scanner = find_index_for_scan(filter);
  if (index_scanner != nullptr) {
    return scan_record_by_index(trx, index_scanner, filter, limit, context, record_reader);
  }

  RC rc = RC::SUCCESS;
  RecordFileScanner scanner;
  rc = scanner.open_scan(*data_buffer_pool_, file_id_, filter);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("failed to open scanner. file id=%d. rc=%d:%s", file_id_, rc, strrc(rc));
    return rc;
  }

  int record_count = 0;
  Record record;
  rc = scanner.get_first_record(&record);
L
Longda 已提交
431
  for (; RC::SUCCESS == rc && record_count < limit; rc = scanner.get_next_record(&record)) {
羽飞's avatar
羽飞 已提交
432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450
    if (trx == nullptr || trx->is_visible(this, &record)) {
      rc = record_reader(&record, context);
      if (rc != RC::SUCCESS) {
        break;
      }
      record_count++;
    }
  }

  if (RC::RECORD_EOF == rc) {
    rc = RC::SUCCESS;
  } else {
    LOG_ERROR("failed to scan record. file id=%d, rc=%d:%s", file_id_, rc, strrc(rc));
  }
  scanner.close_scan();
  return rc;
}

RC Table::scan_record_by_index(Trx *trx, IndexScanner *scanner, ConditionFilter *filter, int limit, void *context,
L
Longda 已提交
451 452
    RC (*record_reader)(Record *, void *))
{
羽飞's avatar
羽飞 已提交
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490
  RC rc = RC::SUCCESS;
  RID rid;
  Record record;
  int record_count = 0;
  while (record_count < limit) {
    rc = scanner->next_entry(&rid);
    if (rc != RC::SUCCESS) {
      if (RC::RECORD_EOF == rc) {
        rc = RC::SUCCESS;
        break;
      }
      LOG_ERROR("Failed to scan table by index. rc=%d:%s", rc, strrc(rc));
      break;
    }

    rc = record_handler_->get_record(&rid, &record);
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to fetch record of rid=%d:%d, rc=%d:%s", rid.page_num, rid.slot_num, rc, strrc(rc));
      break;
    }

    if ((trx == nullptr || trx->is_visible(this, &record)) && (filter == nullptr || filter->filter(record))) {
      rc = record_reader(&record, context);
      if (rc != RC::SUCCESS) {
        LOG_TRACE("Record reader break the table scanning. rc=%d:%s", rc, strrc(rc));
        break;
      }
    }

    record_count++;
  }

  scanner->destroy();
  return rc;
}

class IndexInserter {
public:
L
Longda 已提交
491 492
  explicit IndexInserter(Index *index) : index_(index)
  {}
羽飞's avatar
羽飞 已提交
493

L
Longda 已提交
494 495
  RC insert_index(const Record *record)
  {
羽飞's avatar
羽飞 已提交
496 497
    return index_->insert_entry(record->data, &record->rid);
  }
L
Longda 已提交
498

羽飞's avatar
羽飞 已提交
499
private:
L
Longda 已提交
500
  Index *index_;
羽飞's avatar
羽飞 已提交
501 502
};

L
Longda 已提交
503 504
static RC insert_index_record_reader_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
505 506 507 508
  IndexInserter &inserter = *(IndexInserter *)context;
  return inserter.insert_index(record);
}

L
Longda 已提交
509 510 511 512
RC Table::create_index(Trx *trx, const char *index_name, const char *attribute_name)
{
  if (common::is_blank(index_name) || common::is_blank(attribute_name)) {
    LOG_INFO("Invalid input arguments, table name is %s, index_name is blank or attribute_name is blank", name());
羽飞's avatar
羽飞 已提交
513 514
    return RC::INVALID_ARGUMENT;
  }
L
Longda 已提交
515 516 517 518 519
  if (table_meta_.index(index_name) != nullptr || table_meta_.find_index_by_field((attribute_name))) {
    LOG_INFO("Invalid input arguments, table name is %s, index %s exist or attribute %s exist index",
        name(),
        index_name,
        attribute_name);
羽飞's avatar
羽飞 已提交
520 521 522 523 524
    return RC::SCHEMA_INDEX_EXIST;
  }

  const FieldMeta *field_meta = table_meta_.field(attribute_name);
  if (!field_meta) {
L
Longda 已提交
525
    LOG_INFO("Invalid input arguments, there is no field of %s in table:%s.", attribute_name, name());
羽飞's avatar
羽飞 已提交
526 527 528 529 530 531
    return RC::SCHEMA_FIELD_MISSING;
  }

  IndexMeta new_index_meta;
  RC rc = new_index_meta.init(index_name, *field_meta);
  if (rc != RC::SUCCESS) {
L
Longda 已提交
532
    LOG_INFO("Failed to init IndexMeta in table:%s, index_name:%s, field_name:%s", name(), index_name, attribute_name);
羽飞's avatar
羽飞 已提交
533 534 535 536 537
    return rc;
  }

  // 创建索引相关数据
  BplusTreeIndex *index = new BplusTreeIndex();
L
Longda 已提交
538
  std::string index_file = table_index_file(base_dir_.c_str(), name(), index_name);
羽飞's avatar
羽飞 已提交
539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568
  rc = index->create(index_file.c_str(), new_index_meta, *field_meta);
  if (rc != RC::SUCCESS) {
    delete index;
    LOG_ERROR("Failed to create bplus tree index. file name=%s, rc=%d:%s", index_file.c_str(), rc, strrc(rc));
    return rc;
  }

  // 遍历当前的所有数据,插入这个索引
  IndexInserter index_inserter(index);
  rc = scan_record(trx, nullptr, -1, &index_inserter, insert_index_record_reader_adapter);
  if (rc != RC::SUCCESS) {
    // rollback
    delete index;
    LOG_ERROR("Failed to insert index to all records. table=%s, rc=%d:%s", name(), rc, strrc(rc));
    return rc;
  }
  indexes_.push_back(index);

  TableMeta new_table_meta(table_meta_);
  rc = new_table_meta.add_index(new_index_meta);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to add index (%s) on table (%s). error=%d:%s", index_name, name(), rc, strrc(rc));
    return rc;
  }
  // 创建元数据临时文件
  std::string tmp_file = table_meta_file(base_dir_.c_str(), name()) + ".tmp";
  std::fstream fs;
  fs.open(tmp_file, std::ios_base::out | std::ios_base::binary | std::ios_base::trunc);
  if (!fs.is_open()) {
    LOG_ERROR("Failed to open file for write. file name=%s, errmsg=%s", tmp_file.c_str(), strerror(errno));
L
Longda 已提交
569
    return RC::IOERR;  // 创建索引中途出错,要做还原操作
羽飞's avatar
羽飞 已提交
570 571 572 573 574 575 576 577 578 579 580
  }
  if (new_table_meta.serialize(fs) < 0) {
    LOG_ERROR("Failed to dump new table meta to file: %s. sys err=%d:%s", tmp_file.c_str(), errno, strerror(errno));
    return RC::IOERR;
  }
  fs.close();

  // 覆盖原始元数据文件
  std::string meta_file = table_meta_file(base_dir_.c_str(), name());
  int ret = rename(tmp_file.c_str(), meta_file.c_str());
  if (ret != 0) {
L
Longda 已提交
581 582 583 584 585 586 587 588
    LOG_ERROR("Failed to rename tmp meta file (%s) to normal meta file (%s) while creating index (%s) on table (%s). "
              "system error=%d:%s",
        tmp_file.c_str(),
        meta_file.c_str(),
        index_name,
        name(),
        errno,
        strerror(errno));
羽飞's avatar
羽飞 已提交
589 590 591 592 593
    return RC::IOERR;
  }

  table_meta_.swap(new_table_meta);

L
Longda 已提交
594
  LOG_INFO("Successfully added a new index (%s) on the table (%s)", index_name, name());
羽飞's avatar
羽飞 已提交
595 596 597 598

  return rc;
}

L
Longda 已提交
599 600 601
RC Table::update_record(Trx *trx, const char *attribute_name, const Value *value, int condition_num,
    const Condition conditions[], int *updated_count)
{
羽飞's avatar
羽飞 已提交
602 603 604 605 606
  return RC::GENERIC_ERROR;
}

class RecordDeleter {
public:
L
Longda 已提交
607 608
  RecordDeleter(Table &table, Trx *trx) : table_(table), trx_(trx)
  {}
羽飞's avatar
羽飞 已提交
609

L
Longda 已提交
610 611
  RC delete_record(Record *record)
  {
羽飞's avatar
羽飞 已提交
612 613 614 615 616 617 618 619
    RC rc = RC::SUCCESS;
    rc = table_.delete_record(trx_, record);
    if (rc == RC::SUCCESS) {
      deleted_count_++;
    }
    return rc;
  }

L
Longda 已提交
620 621
  int deleted_count() const
  {
羽飞's avatar
羽飞 已提交
622 623 624 625
    return deleted_count_;
  }

private:
L
Longda 已提交
626
  Table &table_;
羽飞's avatar
羽飞 已提交
627 628 629 630
  Trx *trx_;
  int deleted_count_ = 0;
};

L
Longda 已提交
631 632
static RC record_reader_delete_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
633 634 635 636
  RecordDeleter &record_deleter = *(RecordDeleter *)context;
  return record_deleter.delete_record(record);
}

L
Longda 已提交
637 638
RC Table::delete_record(Trx *trx, ConditionFilter *filter, int *deleted_count)
{
羽飞's avatar
羽飞 已提交
639 640 641 642 643 644 645 646
  RecordDeleter deleter(*this, trx);
  RC rc = scan_record(trx, filter, -1, &deleter, record_reader_delete_adapter);
  if (deleted_count != nullptr) {
    *deleted_count = deleter.deleted_count();
  }
  return rc;
}

L
Longda 已提交
647 648
RC Table::delete_record(Trx *trx, Record *record)
{
羽飞's avatar
羽飞 已提交
649 650 651 652
  RC rc = RC::SUCCESS;
  if (trx != nullptr) {
    rc = trx->delete_record(this, record);
  } else {
L
Longda 已提交
653
    rc = delete_entry_of_indexes(record->data, record->rid, false);  // 重复代码 refer to commit_delete
羽飞's avatar
羽飞 已提交
654 655
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to delete indexes of record (rid=%d.%d). rc=%d:%s",
L
Longda 已提交
656 657 658 659
          record->rid.page_num,
          record->rid.slot_num,
          rc,
          strrc(rc));
羽飞's avatar
羽飞 已提交
660 661 662 663 664 665 666
    } else {
      rc = record_handler_->delete_record(&record->rid);
    }
  }
  return rc;
}

L
Longda 已提交
667 668
RC Table::commit_delete(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
669 670 671 672 673 674 675 676 677
  RC rc = RC::SUCCESS;
  Record record;
  rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
    return rc;
  }
  rc = delete_entry_of_indexes(record.data, record.rid, false);
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to delete indexes of record(rid=%d.%d). rc=%d:%s",
L
Longda 已提交
678 679 680 681
        rid.page_num,
        rid.slot_num,
        rc,
        strrc(rc));  // panic?
羽飞's avatar
羽飞 已提交
682 683 684 685 686 687 688 689 690 691
  }

  rc = record_handler_->delete_record(&rid);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  return rc;
}

L
Longda 已提交
692 693
RC Table::rollback_delete(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
694 695 696 697 698 699 700
  RC rc = RC::SUCCESS;
  Record record;
  rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
    return rc;
  }

L
Longda 已提交
701
  return trx->rollback_delete(this, record);  // update record in place
羽飞's avatar
羽飞 已提交
702 703
}

L
Longda 已提交
704 705
RC Table::insert_entry_of_indexes(const char *record, const RID &rid)
{
羽飞's avatar
羽飞 已提交
706 707 708 709 710 711 712 713 714 715
  RC rc = RC::SUCCESS;
  for (Index *index : indexes_) {
    rc = index->insert_entry(record, &rid);
    if (rc != RC::SUCCESS) {
      break;
    }
  }
  return rc;
}

L
Longda 已提交
716 717
RC Table::delete_entry_of_indexes(const char *record, const RID &rid, bool error_on_not_exists)
{
羽飞's avatar
羽飞 已提交
718 719 720 721 722 723 724 725 726 727 728 729
  RC rc = RC::SUCCESS;
  for (Index *index : indexes_) {
    rc = index->delete_entry(record, &rid);
    if (rc != RC::SUCCESS) {
      if (rc != RC::RECORD_INVALID_KEY || !error_on_not_exists) {
        break;
      }
    }
  }
  return rc;
}

L
Longda 已提交
730 731 732
Index *Table::find_index(const char *index_name) const
{
  for (Index *index : indexes_) {
羽飞's avatar
羽飞 已提交
733 734 735 736 737 738 739
    if (0 == strcmp(index->index_meta().name(), index_name)) {
      return index;
    }
  }
  return nullptr;
}

L
Longda 已提交
740 741
IndexScanner *Table::find_index_for_scan(const DefaultConditionFilter &filter)
{
羽飞's avatar
羽飞 已提交
742 743 744 745 746 747 748 749 750 751 752 753 754 755 756
  const ConDesc *field_cond_desc = nullptr;
  const ConDesc *value_cond_desc = nullptr;
  if (filter.left().is_attr && !filter.right().is_attr) {
    field_cond_desc = &filter.left();
    value_cond_desc = &filter.right();
  } else if (filter.right().is_attr && !filter.left().is_attr) {
    field_cond_desc = &filter.right();
    value_cond_desc = &filter.left();
  }
  if (field_cond_desc == nullptr || value_cond_desc == nullptr) {
    return nullptr;
  }

  const FieldMeta *field_meta = table_meta_.find_field_by_offset(field_cond_desc->attr_offset);
  if (nullptr == field_meta) {
L
Longda 已提交
757
    LOG_PANIC("Cannot find field by offset %d. table=%s", field_cond_desc->attr_offset, name());
羽飞's avatar
羽飞 已提交
758 759 760 761 762 763 764 765 766 767 768 769 770
    return nullptr;
  }

  const IndexMeta *index_meta = table_meta_.find_index_by_field(field_meta->name());
  if (nullptr == index_meta) {
    return nullptr;
  }

  Index *index = find_index(index_meta->name());
  if (nullptr == index) {
    return nullptr;
  }

羽飞's avatar
羽飞 已提交
771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807
  const char *left_key = nullptr;
  const char *right_key = nullptr;
  bool left_inclusive = false;
  bool right_inclusive = false;
  switch (filter.comp_op()) {
  case EQUAL_TO: {
    left_key = (const char *)value_cond_desc->value;
    right_key = (const char *)value_cond_desc->value;
    left_inclusive = true;
    right_inclusive = true;
  }
    break;
  case LESS_EQUAL: {
    right_key = (const char *)value_cond_desc->value;
    right_inclusive = true;
  }
    break;
  case GREAT_EQUAL: {
    left_key = (const char *)value_cond_desc->value;
    left_inclusive = true;
  }
    break;
  case LESS_THAN: {
    right_key = (const char *)value_cond_desc->value;
    right_inclusive = false;
  }
    break;
  case GREAT_THAN: {
    left_key = (const char *)value_cond_desc->value;
    left_inclusive = false;
  }
    break;
  default: {
    return nullptr;
  }
  }
  return index->create_scanner(left_key, left_inclusive, right_key, right_inclusive);
羽飞's avatar
羽飞 已提交
808 809
}

L
Longda 已提交
810 811
IndexScanner *Table::find_index_for_scan(const ConditionFilter *filter)
{
羽飞's avatar
羽飞 已提交
812 813 814 815 816 817 818 819 820 821 822 823 824 825
  if (nullptr == filter) {
    return nullptr;
  }

  // remove dynamic_cast
  const DefaultConditionFilter *default_condition_filter = dynamic_cast<const DefaultConditionFilter *>(filter);
  if (default_condition_filter != nullptr) {
    return find_index_for_scan(*default_condition_filter);
  }

  const CompositeConditionFilter *composite_condition_filter = dynamic_cast<const CompositeConditionFilter *>(filter);
  if (composite_condition_filter != nullptr) {
    int filter_num = composite_condition_filter->filter_num();
    for (int i = 0; i < filter_num; i++) {
L
Longda 已提交
826
      IndexScanner *scanner = find_index_for_scan(&composite_condition_filter->filter(i));
羽飞's avatar
羽飞 已提交
827
      if (scanner != nullptr) {
L
Longda 已提交
828
        return scanner;  // 可以找到一个最优的,比如比较符号是=
羽飞's avatar
羽飞 已提交
829 830 831 832 833 834
      }
    }
  }
  return nullptr;
}

L
Longda 已提交
835 836 837
RC Table::sync()
{
  RC rc = data_buffer_pool_->purge_all_pages(file_id_);
羽飞's avatar
羽飞 已提交
838 839 840 841 842
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to flush table's data pages. table=%s, rc=%d:%s", name(), rc, strrc(rc));
    return rc;
  }

L
Longda 已提交
843
  for (Index *index : indexes_) {
羽飞's avatar
羽飞 已提交
844 845 846
    rc = index->sync();
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to flush index's pages. table=%s, index=%s, rc=%d:%s",
L
Longda 已提交
847 848 849 850
          name(),
          index->index_meta().name(),
          rc,
          strrc(rc));
羽飞's avatar
羽飞 已提交
851 852 853 854 855 856
      return rc;
    }
  }
  LOG_INFO("Sync table over. table=%s", name());
  return rc;
}