Skip to content

Commit

Permalink
Merge pull request cmu-db#1383 from ksaito7/catalog_update
Browse files Browse the repository at this point in the history 
Catalog updates for checkpoints
  • Loading branch information
@tli2
tli2 authored Jun 18, 2018
2 parents cc3346b + 8564d4f commit 931fa92
Show file tree
Hide file tree
Showing 28 changed files with 390 additions and 149 deletions.
76 changes: 54 additions & 22 deletions src/catalog/catalog.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -169,22 +169,22 @@ void Catalog::BootstrapSystemCatalogs(storage::Database *database,
// pg_database record is shared across different databases
system_catalogs->GetTableCatalog()->InsertTable(
DATABASE_CATALOG_OID, DATABASE_CATALOG_NAME, CATALOG_SCHEMA_NAME,
CATALOG_DATABASE_OID, pool_.get(), txn);
CATALOG_DATABASE_OID, ROW_STORE_LAYOUT_OID, pool_.get(), txn);
system_catalogs->GetTableCatalog()->InsertTable(
SCHEMA_CATALOG_OID, SCHEMA_CATALOG_NAME, CATALOG_SCHEMA_NAME,
database_oid, pool_.get(), txn);
database_oid, ROW_STORE_LAYOUT_OID, pool_.get(), txn);
system_catalogs->GetTableCatalog()->InsertTable(
TABLE_CATALOG_OID, TABLE_CATALOG_NAME, CATALOG_SCHEMA_NAME, database_oid,
pool_.get(), txn);
ROW_STORE_LAYOUT_OID, pool_.get(), txn);
system_catalogs->GetTableCatalog()->InsertTable(
INDEX_CATALOG_OID, INDEX_CATALOG_NAME, CATALOG_SCHEMA_NAME, database_oid,
pool_.get(), txn);
ROW_STORE_LAYOUT_OID, pool_.get(), txn);
system_catalogs->GetTableCatalog()->InsertTable(
COLUMN_CATALOG_OID, COLUMN_CATALOG_NAME, CATALOG_SCHEMA_NAME,
database_oid, pool_.get(), txn);
database_oid, ROW_STORE_LAYOUT_OID, pool_.get(), txn);
system_catalogs->GetTableCatalog()->InsertTable(
LAYOUT_CATALOG_OID, LAYOUT_CATALOG_NAME, CATALOG_SCHEMA_NAME,
database_oid, pool_.get(), txn);
database_oid, ROW_STORE_LAYOUT_OID, pool_.get(), txn);
}

void Catalog::Bootstrap() {
Expand All @@ -208,6 +208,12 @@ void Catalog::Bootstrap() {

InitializeLanguages();
InitializeFunctions();

// Reset oid of each catalog to avoid collisions between catalog
// values added by system and users when checkpoint recovery.
DatabaseCatalog::GetInstance()->UpdateOid(OID_FOR_USER_OFFSET);
LanguageCatalog::GetInstance().UpdateOid(OID_FOR_USER_OFFSET);
ProcCatalog::GetInstance().UpdateOid(OID_FOR_USER_OFFSET);
}

//===----------------------------------------------------------------------===//
Expand Down Expand Up @@ -362,13 +368,14 @@ ResultType Catalog::CreateTable(const std::string &database_name,

// Update pg_table with table info
pg_table->InsertTable(table_oid, table_name, schema_name,
database_object->GetDatabaseOid(), pool_.get(), txn);
database_object->GetDatabaseOid(),
table->GetDefaultLayout()->GetOid(), pool_.get(), txn);
oid_t column_id = 0;
for (const auto &column : table->GetSchema()->GetColumns()) {
pg_attribute->InsertColumn(table_oid, column.GetName(), column_id,
column.GetOffset(), column.GetType(),
column.IsInlined(), column.GetConstraints(),
pool_.get(), txn);
column.GetLength(), column.IsInlined(),
column.GetConstraints(), pool_.get(), txn);

// Create index on unique single column
if (column.IsUnique()) {
Expand All @@ -383,6 +390,15 @@ ResultType Catalog::CreateTable(const std::string &database_name,
}
CreatePrimaryIndex(database_object->GetDatabaseOid(), table_oid, schema_name,
txn);

// Create layout as default layout
auto pg_layout =
catalog_map_[database_object->GetDatabaseOid()]->GetLayoutCatalog();
auto default_layout = table->GetDefaultLayout();
if (!pg_layout->InsertLayout(table_oid, default_layout, pool_.get(), txn))
throw CatalogException("Failed to create a new layout for table "
+ table_name);

return ResultType::SUCCESS;
}

Expand Down Expand Up @@ -573,15 +589,15 @@ std::shared_ptr<const storage::Layout> Catalog::CreateLayout(
// Ensure that the new layout
PELOTON_ASSERT(layout_oid < INVALID_OID);
auto new_layout = std::shared_ptr<const storage::Layout>(
new const storage::Layout(column_map, layout_oid));
new const storage::Layout(column_map, column_map.size(), layout_oid));

// Add the layout the pg_layout table
auto pg_layout = catalog_map_[database_oid]->GetLayoutCatalog();
bool result =
pg_layout->InsertLayout(table_oid, new_layout, pool_.get(), txn);
if (!result) {
LOG_ERROR("Failed to create a new layout for table %u", table_oid);
return nullptr;
if (pg_layout->GetLayoutWithOid(table_oid, new_layout->GetOid(), txn)
== nullptr &&
!pg_layout->InsertLayout(table_oid, new_layout, pool_.get(), txn)) {
LOG_ERROR("Failed to create a new layout for table %u", table_oid);
return nullptr;
}
return new_layout;
}
Expand All @@ -596,6 +612,10 @@ std::shared_ptr<const storage::Layout> Catalog::CreateDefaultLayout(
auto database = storage_manager->GetDatabaseWithOid(database_oid);
auto table = database->GetTableWithOid(table_oid);
table->SetDefaultLayout(new_layout);

// update table catalog
catalog_map_[database_oid]->GetTableCatalog()
->UpdateDefaultLayoutOid(new_layout->GetOid(), table_oid, txn);
}
return new_layout;
}
Expand Down Expand Up @@ -791,7 +811,7 @@ ResultType Catalog::DropIndex(oid_t database_oid, oid_t index_oid,
// find index catalog object by looking up pg_index or read from cache using
// index_oid
auto pg_index = catalog_map_[database_oid]->GetIndexCatalog();
auto index_object = pg_index->GetIndexObject(index_oid, txn);
auto index_object = pg_index->GetIndexObject(database_oid, index_oid, txn);
if (index_object == nullptr) {
throw CatalogException("Can't find index " + std::to_string(index_oid) +
" to drop");
Expand All @@ -801,7 +821,7 @@ ResultType Catalog::DropIndex(oid_t database_oid, oid_t index_oid,
auto table = storage_manager->GetTableWithOid(database_oid,
index_object->GetTableOid());
// drop record in pg_index
pg_index->DeleteIndex(index_oid, txn);
pg_index->DeleteIndex(database_oid, index_oid, txn);
LOG_TRACE("Successfully drop index %d for table %s", index_oid,
table->GetName().c_str());

Expand All @@ -822,17 +842,29 @@ ResultType Catalog::DropLayout(oid_t database_oid, oid_t table_oid,
auto table = database->GetTableWithOid(table_oid);
auto default_layout = table->GetDefaultLayout();

if (default_layout.GetOid() == layout_oid) {
table->ResetDefaultLayout();
}

auto pg_layout = catalog_map_[database_oid]->GetLayoutCatalog();
if (!pg_layout->DeleteLayout(table_oid, layout_oid, txn)) {
auto layout = table->GetDefaultLayout();
LOG_DEBUG("Layout delete failed. Default layout id: %u", layout.GetOid());
LOG_DEBUG("Layout delete failed. Default layout id: %u", layout->GetOid());
return ResultType::FAILURE;
}

if (default_layout->GetOid() == layout_oid) {
table->ResetDefaultLayout();
auto new_default_layout = table->GetDefaultLayout();
if (pg_layout->GetLayoutWithOid(table_oid, new_default_layout->GetOid(),
txn) == nullptr &&
!pg_layout->InsertLayout(table_oid, new_default_layout,
pool_.get(), txn)) {
LOG_DEBUG("Failed to create a new layout for table %d", table_oid);
return ResultType::FAILURE;
}

// update table catalog
catalog_map_[database_oid]->GetTableCatalog()
->UpdateDefaultLayoutOid(new_default_layout->GetOid(), table_oid, txn);
}

return ResultType::SUCCESS;
}

Expand Down
39 changes: 17 additions & 22 deletions src/catalog/catalog_cache.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,13 +116,11 @@ std::shared_ptr<DatabaseCatalogObject> CatalogCache::GetDatabaseObject(
* @return table catalog object; if not found return null
*/
std::shared_ptr<TableCatalogObject> CatalogCache::GetCachedTableObject(
oid_t table_oid) {
for (auto it = database_objects_cache.begin();
it != database_objects_cache.end(); ++it) {
auto database_object = it->second;
auto table_object = database_object->GetTableObject(table_oid, true);
if (table_object) return table_object;
}
oid_t database_oid, oid_t table_oid) {
auto database_object = GetDatabaseObject(database_oid);
if (database_object == nullptr) return nullptr;
auto table_object = database_object->GetTableObject(table_oid, true);
if (table_object) return table_object;
return nullptr;
}

Expand All @@ -131,13 +129,11 @@ std::shared_ptr<TableCatalogObject> CatalogCache::GetCachedTableObject(
* @return index catalog object; if not found return null
*/
std::shared_ptr<IndexCatalogObject> CatalogCache::GetCachedIndexObject(
oid_t index_oid) {
for (auto it = database_objects_cache.begin();
it != database_objects_cache.end(); ++it) {
auto database_object = it->second;
auto index_object = database_object->GetCachedIndexObject(index_oid);
if (index_object) return index_object;
}
oid_t database_oid, oid_t index_oid) {
auto database_object = GetDatabaseObject(database_oid);
if (database_object == nullptr) return nullptr;
auto index_object = database_object->GetCachedIndexObject(index_oid);
if (index_object) return index_object;
return nullptr;
}

Expand All @@ -146,14 +142,13 @@ std::shared_ptr<IndexCatalogObject> CatalogCache::GetCachedIndexObject(
* @return index catalog object; if not found return null
*/
std::shared_ptr<IndexCatalogObject> CatalogCache::GetCachedIndexObject(
const std::string &index_name, const std::string &schema_name) {
for (auto it = database_objects_cache.begin();
it != database_objects_cache.end(); ++it) {
auto database_object = it->second;
auto index_object =
database_object->GetCachedIndexObject(index_name, schema_name);
if (index_object) return index_object;
}
const std::string &database_name, const std::string &index_name,
const std::string &schema_name) {
auto database_object = GetDatabaseObject(database_name);
if (database_object == nullptr) return nullptr;
auto index_object =
database_object->GetCachedIndexObject(index_name, schema_name);
if (index_object) return index_object;
return nullptr;
}

Expand Down
38 changes: 25 additions & 13 deletions src/catalog/column_catalog.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,16 @@ ColumnCatalogObject::ColumnCatalogObject(executor::LogicalTile *tile,
column_name(tile->GetValue(tupleId, ColumnCatalog::ColumnId::COLUMN_NAME)
.ToString()),
column_id(tile->GetValue(tupleId, ColumnCatalog::ColumnId::COLUMN_ID)
.GetAs<uint32_t>()),
.GetAs<oid_t>()),
column_offset(
tile->GetValue(tupleId, ColumnCatalog::ColumnId::COLUMN_OFFSET)
.GetAs<uint32_t>()),
.GetAs<oid_t>()),
column_type(StringToTypeId(
tile->GetValue(tupleId, ColumnCatalog::ColumnId::COLUMN_TYPE)
.ToString())),
column_length(
tile->GetValue(tupleId, ColumnCatalog::ColumnId::COLUMN_LENGTH)
.GetAs<uint32_t>()),
is_inlined(tile->GetValue(tupleId, ColumnCatalog::ColumnId::IS_INLINED)
.GetAs<bool>()),
is_primary(tile->GetValue(tupleId, ColumnCatalog::ColumnId::IS_PRIMARY)
Expand All @@ -62,8 +65,8 @@ ColumnCatalog::ColumnCatalog(storage::Database *pg_catalog,
uint32_t column_id = 0;
for (auto column : catalog_table_->GetSchema()->GetColumns()) {
InsertColumn(COLUMN_CATALOG_OID, column.GetName(), column_id,
column.GetOffset(), column.GetType(), column.IsInlined(),
column.GetConstraints(), pool, txn);
column.GetOffset(), column.GetType(), column.GetLength(),
column.IsInlined(), column.GetConstraints(), pool, txn);
column_id++;
}
}
Expand Down Expand Up @@ -109,6 +112,12 @@ std::unique_ptr<catalog::Schema> ColumnCatalog::InitializeSchema() {
column_type_column.AddConstraint(
catalog::Constraint(ConstraintType::NOTNULL, not_null_constraint_name));

auto column_length_column = catalog::Column(
type::TypeId::INTEGER, type::Type::GetTypeSize(type::TypeId::INTEGER),
"column_length", true);
column_length_column.AddConstraint(
catalog::Constraint(ConstraintType::NOTNULL, not_null_constraint_name));

auto is_inlined_column = catalog::Column(
type::TypeId::BOOLEAN, type::Type::GetTypeSize(type::TypeId::BOOLEAN),
"is_inlined", true);
Expand All @@ -129,16 +138,17 @@ std::unique_ptr<catalog::Schema> ColumnCatalog::InitializeSchema() {

std::unique_ptr<catalog::Schema> column_catalog_schema(new catalog::Schema(
{table_id_column, column_name_column, column_id_column,
column_offset_column, column_type_column, is_inlined_column,
is_primary_column, is_not_null_column}));
column_offset_column, column_type_column, column_length_column,
is_inlined_column, is_primary_column, is_not_null_column}));

return column_catalog_schema;
}

bool ColumnCatalog::InsertColumn(oid_t table_oid,
const std::string &column_name,
oid_t column_id, oid_t column_offset,
type::TypeId column_type, bool is_inlined,
type::TypeId column_type, size_t column_length,
bool is_inlined,
const std::vector<Constraint> &constraints,
type::AbstractPool *pool,
concurrency::TransactionContext *txn) {
Expand All @@ -152,7 +162,8 @@ bool ColumnCatalog::InsertColumn(oid_t table_oid,
auto val3 = type::ValueFactory::GetIntegerValue(column_offset);
auto val4 =
type::ValueFactory::GetVarcharValue(TypeIdToString(column_type), nullptr);
auto val5 = type::ValueFactory::GetBooleanValue(is_inlined);
auto val5 = type::ValueFactory::GetIntegerValue(column_length);
auto val6 = type::ValueFactory::GetBooleanValue(is_inlined);
bool is_primary = false, is_not_null = false;
for (auto constraint : constraints) {
if (constraint.GetType() == ConstraintType::PRIMARY) {
Expand All @@ -162,17 +173,18 @@ bool ColumnCatalog::InsertColumn(oid_t table_oid,
is_not_null = true;
}
}
auto val6 = type::ValueFactory::GetBooleanValue(is_primary);
auto val7 = type::ValueFactory::GetBooleanValue(is_not_null);
auto val7 = type::ValueFactory::GetBooleanValue(is_primary);
auto val8 = type::ValueFactory::GetBooleanValue(is_not_null);

tuple->SetValue(ColumnId::TABLE_OID, val0, pool);
tuple->SetValue(ColumnId::COLUMN_NAME, val1, pool);
tuple->SetValue(ColumnId::COLUMN_ID, val2, pool);
tuple->SetValue(ColumnId::COLUMN_OFFSET, val3, pool);
tuple->SetValue(ColumnId::COLUMN_TYPE, val4, pool);
tuple->SetValue(ColumnId::IS_INLINED, val5, pool);
tuple->SetValue(ColumnId::IS_PRIMARY, val6, pool);
tuple->SetValue(ColumnId::IS_NOT_NULL, val7, pool);
tuple->SetValue(ColumnId::COLUMN_LENGTH, val5, pool);
tuple->SetValue(ColumnId::IS_INLINED, val6, pool);
tuple->SetValue(ColumnId::IS_PRIMARY, val7, pool);
tuple->SetValue(ColumnId::IS_NOT_NULL, val8, pool);

// Insert the tuple
return InsertTuple(std::move(tuple), txn);
Expand Down
20 changes: 12 additions & 8 deletions src/catalog/index_catalog.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -167,29 +167,32 @@ bool IndexCatalog::InsertIndex(oid_t index_oid, const std::string &index_name,
return InsertTuple(std::move(tuple), txn);
}

bool IndexCatalog::DeleteIndex(oid_t index_oid,
bool IndexCatalog::DeleteIndex(oid_t database_oid, oid_t index_oid,
concurrency::TransactionContext *txn) {
oid_t index_offset = IndexId::PRIMARY_KEY; // Index of index_oid
std::vector<type::Value> values;
values.push_back(type::ValueFactory::GetIntegerValue(index_oid).Copy());

auto index_object = txn->catalog_cache.GetCachedIndexObject(index_oid);
auto index_object = txn->catalog_cache.GetCachedIndexObject(database_oid,
index_oid);
if (index_object) {
auto table_object =
txn->catalog_cache.GetCachedTableObject(index_object->GetTableOid());
txn->catalog_cache.GetCachedTableObject(database_oid,
index_object->GetTableOid());
table_object->EvictAllIndexObjects();
}

return DeleteWithIndexScan(index_offset, values, txn);
}

std::shared_ptr<IndexCatalogObject> IndexCatalog::GetIndexObject(
oid_t index_oid, concurrency::TransactionContext *txn) {
oid_t database_oid, oid_t index_oid, concurrency::TransactionContext *txn) {
if (txn == nullptr) {
throw CatalogException("Transaction is invalid!");
}
// try get from cache
auto index_object = txn->catalog_cache.GetCachedIndexObject(index_oid);
auto index_object = txn->catalog_cache.GetCachedIndexObject(database_oid,
index_oid);
if (index_object) {
return index_object;
}
Expand Down Expand Up @@ -224,14 +227,15 @@ std::shared_ptr<IndexCatalogObject> IndexCatalog::GetIndexObject(
}

std::shared_ptr<IndexCatalogObject> IndexCatalog::GetIndexObject(
const std::string &index_name, const std::string &schema_name,
concurrency::TransactionContext *txn) {
const std::string &database_name, const std::string &index_name,
const std::string &schema_name, concurrency::TransactionContext *txn) {
if (txn == nullptr) {
throw CatalogException("Transaction is invalid!");
}
// try get from cache
auto index_object =
txn->catalog_cache.GetCachedIndexObject(index_name, schema_name);
txn->catalog_cache.GetCachedIndexObject(database_name, index_name,
schema_name);
if (index_object) {
return index_object;
}
Expand Down
Loading

0 comments on commit 931fa92

Please sign in to comment.