table.cpp 24.1 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 34 35 36 37
Table::~Table()
{
  if (record_handler_ != nullptr) {
    delete record_handler_;
    record_handler_ = nullptr;
  }
羽飞's avatar
羽飞 已提交
38

羽飞's avatar
羽飞 已提交
39 40
  if (data_buffer_pool_ != nullptr) {
    data_buffer_pool_->close_file();
羽飞's avatar
羽飞 已提交
41 42 43
    data_buffer_pool_ = nullptr;
  }

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

羽飞's avatar
羽飞 已提交
50 51 52
  LOG_INFO("Table has been closed: %s", name());
}

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

L
Longda 已提交
57
  if (common::is_blank(name)) {
羽飞's avatar
羽飞 已提交
58 59 60 61 62 63
    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 已提交
64
    LOG_WARN("Invalid arguments. table_name=%s, attribute_count=%d, attributes=%p", name, attribute_count, attributes);
羽飞's avatar
羽飞 已提交
65 66 67 68 69 70 71 72
    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 已提交
73
  if (fd < 0) {
羽飞's avatar
羽飞 已提交
74
    if (EEXIST == errno) {
L
Longda 已提交
75
      LOG_ERROR("Failed to create table file, it has been created. %s, EEXIST, %s", path, strerror(errno));
羽飞's avatar
羽飞 已提交
76 77
      return RC::SCHEMA_TABLE_EXIST;
    }
L
Longda 已提交
78
    LOG_ERROR("Create table file failed. filename=%s, errmsg=%d:%s", path, errno, strerror(errno));
羽飞's avatar
羽飞 已提交
79 80 81 82 83 84 85 86
    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 已提交
87
    return rc;  // delete table file
羽飞's avatar
羽飞 已提交
88 89 90 91 92 93 94 95 96 97 98 99 100
  }

  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 已提交
101
  std::string data_file = table_data_file(base_dir, name);
羽飞's avatar
羽飞 已提交
102 103
  BufferPoolManager &bpm = BufferPoolManager::instance();
  rc = bpm.create_file(data_file.c_str());
羽飞's avatar
羽飞 已提交
104 105 106 107 108 109
  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 已提交
110 111 112 113 114
  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
羽飞 已提交
115 116 117 118 119 120

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

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

  // 加载数据文件
  RC rc = init_record_handler(base_dir);
L
Longda 已提交
140 141 142 143 144
  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
羽飞 已提交
145 146 147 148 149 150 151 152

  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 已提交
153 154 155 156 157 158
      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
羽飞 已提交
159 160 161 162
      return RC::GENERIC_ERROR;
    }

    BplusTreeIndex *index = new BplusTreeIndex();
L
Longda 已提交
163
    std::string index_file = table_index_file(base_dir, name(), index_meta->name());
羽飞's avatar
羽飞 已提交
164 165 166 167
    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 已提交
168 169 170 171 172 173 174
          name(),
          index_meta->name(),
          index_file.c_str(),
          rc,
          strrc(rc));
      // skip cleanup
      //  do all cleanup action in destructive Table function.
羽飞's avatar
羽飞 已提交
175 176 177 178 179 180 181
      return rc;
    }
    indexes_.push_back(index);
  }
  return rc;
}

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

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

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

  Record record;
  RC rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
L
Longda 已提交
200
    LOG_ERROR("Failed to get record %s: %s", this->name(), rid.to_string().c_str());
羽飞's avatar
羽飞 已提交
201 202 203 204 205 206 207
    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 已提交
208 209 210 211 212
        rid.page_num,
        rid.slot_num,
        rc,
        strrc(rc));
    return rc;
羽飞's avatar
羽飞 已提交
213
  }
L
Longda 已提交
214 215

  rc = record_handler_->delete_record(&rid);
羽飞's avatar
羽飞 已提交
216 217 218
  return rc;
}

L
Longda 已提交
219 220
RC Table::insert_record(Trx *trx, Record *record)
{
羽飞's avatar
羽飞 已提交
221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238
  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 已提交
239 240 241 242
        LOG_ERROR("Failed to rollback record data when insert index entries failed. table name=%s, rc=%d:%s",
            name(),
            rc2,
            strrc(rc2));
羽飞's avatar
羽飞 已提交
243 244 245 246 247 248 249 250 251
      }
      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 已提交
252 253 254 255
      LOG_ERROR("Failed to rollback index data when insert index entries failed. table name=%s, rc=%d:%s",
          name(),
          rc2,
          strrc(rc2));
羽飞's avatar
羽飞 已提交
256 257 258 259
    }
    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 已提交
260 261 262
          name(),
          rc2,
          strrc(rc2));
羽飞's avatar
羽飞 已提交
263 264 265 266 267
    }
    return rc;
  }
  return rc;
}
L
Longda 已提交
268 269 270 271
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
羽飞 已提交
272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289
    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 已提交
290 291
const char *Table::name() const
{
羽飞's avatar
羽飞 已提交
292 293 294
  return table_meta_.name();
}

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

L
Longda 已提交
300 301
RC Table::make_record(int value_num, const Value *values, char *&record_out)
{
羽飞's avatar
羽飞 已提交
302 303
  // 检查字段类型是否一致
  if (value_num + table_meta_.sys_field_num() != table_meta_.field_num()) {
L
Longda 已提交
304
    LOG_WARN("Input values don't match the table's schema, table name:%s", table_meta_.name());
羽飞's avatar
羽飞 已提交
305 306 307 308 309 310 311 312
    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 已提交
313 314 315 316 317
      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
羽飞 已提交
318 319 320 321 322 323
      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
    }
  }

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

  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 已提交
336 337 338
RC Table::init_record_handler(const char *base_dir)
{
  std::string data_file = table_data_file(base_dir, table_meta_.name());
羽飞's avatar
羽飞 已提交
339

羽飞's avatar
羽飞 已提交
340
  RC rc = BufferPoolManager::instance().open_file(data_file.c_str(), data_buffer_pool_);
羽飞's avatar
羽飞 已提交
341
  if (rc != RC::SUCCESS) {
L
Longda 已提交
342
    LOG_ERROR("Failed to open disk buffer pool for file:%s. rc=%d:%s", data_file.c_str(), rc, strrc(rc));
羽飞's avatar
羽飞 已提交
343 344 345 346
    return rc;
  }

  record_handler_ = new RecordFileHandler();
羽飞's avatar
羽飞 已提交
347
  rc = record_handler_->init(data_buffer_pool_);
羽飞's avatar
羽飞 已提交
348 349
  if (rc != RC::SUCCESS) {
    LOG_ERROR("Failed to init record handler. rc=%d:%s", rc, strrc(rc));
羽飞's avatar
羽飞 已提交
350 351
    data_buffer_pool_->close_file();
    data_buffer_pool_ = nullptr;
L
Longda 已提交
352 353
    delete record_handler_;
    record_handler_ = nullptr;
羽飞's avatar
羽飞 已提交
354 355 356 357 358 359 360 361 362 363 364 365
    return rc;
  }

  return rc;
}

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

L
Longda 已提交
369 370
  void consume(const Record *record)
  {
羽飞's avatar
羽飞 已提交
371 372
    record_reader_(record->data, context_);
  }
L
Longda 已提交
373

羽飞's avatar
羽飞 已提交
374 375 376 377
private:
  void (*record_reader_)(const char *, void *);
  void *context_;
};
L
Longda 已提交
378 379 380

static RC scan_record_reader_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
381 382 383 384 385
  RecordReaderScanAdapter &adapter = *(RecordReaderScanAdapter *)context;
  adapter.consume(record);
  return RC::SUCCESS;
}

L
Longda 已提交
386 387 388
RC Table::scan_record(
    Trx *trx, ConditionFilter *filter, int limit, void *context, void (*record_reader)(const char *data, void *context))
{
羽飞's avatar
羽飞 已提交
389 390 391 392
  RecordReaderScanAdapter adapter(record_reader, context);
  return scan_record(trx, filter, limit, (void *)&adapter, scan_record_reader_adapter);
}

L
Longda 已提交
393 394 395
RC Table::scan_record(
    Trx *trx, ConditionFilter *filter, int limit, void *context, RC (*record_reader)(Record *record, void *context))
{
羽飞's avatar
羽飞 已提交
396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414
  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;
羽飞's avatar
羽飞 已提交
415
  rc = scanner.open_scan(*data_buffer_pool_, filter);
羽飞's avatar
羽飞 已提交
416
  if (rc != RC::SUCCESS) {
羽飞's avatar
羽飞 已提交
417
    LOG_ERROR("failed to open scanner. rc=%d:%s", rc, strrc(rc));
羽飞's avatar
羽飞 已提交
418 419 420 421 422 423
    return rc;
  }

  int record_count = 0;
  Record record;
  rc = scanner.get_first_record(&record);
L
Longda 已提交
424
  for (; RC::SUCCESS == rc && record_count < limit; rc = scanner.get_next_record(&record)) {
羽飞's avatar
羽飞 已提交
425 426 427 428 429 430 431 432 433 434 435 436
    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 {
羽飞's avatar
羽飞 已提交
437
    LOG_ERROR("failed to scan record. rc=%d:%s", rc, strrc(rc));
羽飞's avatar
羽飞 已提交
438 439 440 441 442 443
  }
  scanner.close_scan();
  return rc;
}

RC Table::scan_record_by_index(Trx *trx, IndexScanner *scanner, ConditionFilter *filter, int limit, void *context,
L
Longda 已提交
444 445
    RC (*record_reader)(Record *, void *))
{
羽飞's avatar
羽飞 已提交
446 447 448 449 450 451 452 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
  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 已提交
484 485
  explicit IndexInserter(Index *index) : index_(index)
  {}
羽飞's avatar
羽飞 已提交
486

L
Longda 已提交
487 488
  RC insert_index(const Record *record)
  {
羽飞's avatar
羽飞 已提交
489 490
    return index_->insert_entry(record->data, &record->rid);
  }
L
Longda 已提交
491

羽飞's avatar
羽飞 已提交
492
private:
L
Longda 已提交
493
  Index *index_;
羽飞's avatar
羽飞 已提交
494 495
};

L
Longda 已提交
496 497
static RC insert_index_record_reader_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
498 499 500 501
  IndexInserter &inserter = *(IndexInserter *)context;
  return inserter.insert_index(record);
}

L
Longda 已提交
502 503 504 505
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
羽飞 已提交
506 507
    return RC::INVALID_ARGUMENT;
  }
L
Longda 已提交
508 509 510 511 512
  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
羽飞 已提交
513 514 515 516 517
    return RC::SCHEMA_INDEX_EXIST;
  }

  const FieldMeta *field_meta = table_meta_.field(attribute_name);
  if (!field_meta) {
L
Longda 已提交
518
    LOG_INFO("Invalid input arguments, there is no field of %s in table:%s.", attribute_name, name());
羽飞's avatar
羽飞 已提交
519 520 521 522 523 524
    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 已提交
525
    LOG_INFO("Failed to init IndexMeta in table:%s, index_name:%s, field_name:%s", name(), index_name, attribute_name);
羽飞's avatar
羽飞 已提交
526 527 528 529 530
    return rc;
  }

  // 创建索引相关数据
  BplusTreeIndex *index = new BplusTreeIndex();
L
Longda 已提交
531
  std::string index_file = table_index_file(base_dir_.c_str(), name(), index_name);
羽飞's avatar
羽飞 已提交
532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561
  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 已提交
562
    return RC::IOERR;  // 创建索引中途出错,要做还原操作
羽飞's avatar
羽飞 已提交
563 564 565 566 567 568 569 570 571 572 573
  }
  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 已提交
574 575 576 577 578 579 580 581
    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
羽飞 已提交
582 583 584 585 586
    return RC::IOERR;
  }

  table_meta_.swap(new_table_meta);

L
Longda 已提交
587
  LOG_INFO("Successfully added a new index (%s) on the table (%s)", index_name, name());
羽飞's avatar
羽飞 已提交
588 589 590 591

  return rc;
}

L
Longda 已提交
592 593 594
RC Table::update_record(Trx *trx, const char *attribute_name, const Value *value, int condition_num,
    const Condition conditions[], int *updated_count)
{
羽飞's avatar
羽飞 已提交
595 596 597 598 599
  return RC::GENERIC_ERROR;
}

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

L
Longda 已提交
603 604
  RC delete_record(Record *record)
  {
羽飞's avatar
羽飞 已提交
605 606 607 608 609 610 611 612
    RC rc = RC::SUCCESS;
    rc = table_.delete_record(trx_, record);
    if (rc == RC::SUCCESS) {
      deleted_count_++;
    }
    return rc;
  }

L
Longda 已提交
613 614
  int deleted_count() const
  {
羽飞's avatar
羽飞 已提交
615 616 617 618
    return deleted_count_;
  }

private:
L
Longda 已提交
619
  Table &table_;
羽飞's avatar
羽飞 已提交
620 621 622 623
  Trx *trx_;
  int deleted_count_ = 0;
};

L
Longda 已提交
624 625
static RC record_reader_delete_adapter(Record *record, void *context)
{
羽飞's avatar
羽飞 已提交
626 627 628 629
  RecordDeleter &record_deleter = *(RecordDeleter *)context;
  return record_deleter.delete_record(record);
}

L
Longda 已提交
630 631
RC Table::delete_record(Trx *trx, ConditionFilter *filter, int *deleted_count)
{
羽飞's avatar
羽飞 已提交
632 633 634 635 636 637 638 639
  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 已提交
640 641
RC Table::delete_record(Trx *trx, Record *record)
{
羽飞's avatar
羽飞 已提交
642 643 644 645
  RC rc = RC::SUCCESS;
  if (trx != nullptr) {
    rc = trx->delete_record(this, record);
  } else {
L
Longda 已提交
646
    rc = delete_entry_of_indexes(record->data, record->rid, false);  // 重复代码 refer to commit_delete
羽飞's avatar
羽飞 已提交
647 648
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to delete indexes of record (rid=%d.%d). rc=%d:%s",
L
Longda 已提交
649 650 651 652
          record->rid.page_num,
          record->rid.slot_num,
          rc,
          strrc(rc));
羽飞's avatar
羽飞 已提交
653 654 655 656 657 658 659
    } else {
      rc = record_handler_->delete_record(&record->rid);
    }
  }
  return rc;
}

L
Longda 已提交
660 661
RC Table::commit_delete(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
662 663 664 665 666 667 668 669 670
  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 已提交
671 672 673 674
        rid.page_num,
        rid.slot_num,
        rc,
        strrc(rc));  // panic?
羽飞's avatar
羽飞 已提交
675 676 677 678 679 680 681 682 683 684
  }

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

  return rc;
}

L
Longda 已提交
685 686
RC Table::rollback_delete(Trx *trx, const RID &rid)
{
羽飞's avatar
羽飞 已提交
687 688 689 690 691 692 693
  RC rc = RC::SUCCESS;
  Record record;
  rc = record_handler_->get_record(&rid, &record);
  if (rc != RC::SUCCESS) {
    return rc;
  }

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

L
Longda 已提交
697 698
RC Table::insert_entry_of_indexes(const char *record, const RID &rid)
{
羽飞's avatar
羽飞 已提交
699 700 701 702 703 704 705 706 707 708
  RC rc = RC::SUCCESS;
  for (Index *index : indexes_) {
    rc = index->insert_entry(record, &rid);
    if (rc != RC::SUCCESS) {
      break;
    }
  }
  return rc;
}

L
Longda 已提交
709 710
RC Table::delete_entry_of_indexes(const char *record, const RID &rid, bool error_on_not_exists)
{
羽飞's avatar
羽飞 已提交
711 712 713 714 715 716 717 718 719 720 721 722
  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 已提交
723 724 725
Index *Table::find_index(const char *index_name) const
{
  for (Index *index : indexes_) {
羽飞's avatar
羽飞 已提交
726 727 728 729 730 731 732
    if (0 == strcmp(index->index_meta().name(), index_name)) {
      return index;
    }
  }
  return nullptr;
}

L
Longda 已提交
733 734
IndexScanner *Table::find_index_for_scan(const DefaultConditionFilter &filter)
{
羽飞's avatar
羽飞 已提交
735 736 737 738 739 740 741 742 743 744 745 746 747 748 749
  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 已提交
750
    LOG_PANIC("Cannot find field by offset %d. table=%s", field_cond_desc->attr_offset, name());
羽飞's avatar
羽飞 已提交
751 752 753 754 755 756 757 758 759 760 761 762 763
    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
羽飞 已提交
764 765 766 767 768 769 770 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
  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
羽飞 已提交
801 802
}

L
Longda 已提交
803 804
IndexScanner *Table::find_index_for_scan(const ConditionFilter *filter)
{
羽飞's avatar
羽飞 已提交
805 806 807 808 809 810 811 812 813 814 815 816 817 818
  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 已提交
819
      IndexScanner *scanner = find_index_for_scan(&composite_condition_filter->filter(i));
羽飞's avatar
羽飞 已提交
820
      if (scanner != nullptr) {
L
Longda 已提交
821
        return scanner;  // 可以找到一个最优的,比如比较符号是=
羽飞's avatar
羽飞 已提交
822 823 824 825 826 827
      }
    }
  }
  return nullptr;
}

L
Longda 已提交
828 829
RC Table::sync()
{
羽飞's avatar
羽飞 已提交
830
  RC rc = data_buffer_pool_->purge_all_pages();
羽飞's avatar
羽飞 已提交
831 832 833 834 835
  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 已提交
836
  for (Index *index : indexes_) {
羽飞's avatar
羽飞 已提交
837 838 839
    rc = index->sync();
    if (rc != RC::SUCCESS) {
      LOG_ERROR("Failed to flush index's pages. table=%s, index=%s, rc=%d:%s",
L
Longda 已提交
840 841 842 843
          name(),
          index->index_meta().name(),
          rc,
          strrc(rc));
羽飞's avatar
羽飞 已提交
844 845 846 847 848 849
      return rc;
    }
  }
  LOG_INFO("Sync table over. table=%s", name());
  return rc;
}