Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add show buffer command #1299

Merged
merged 3 commits into from
Jun 6, 2024
Merged

Add show buffer command #1299

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
48d2d37
Add show buffer command
JinHai-CN Jun 6, 2024
16dc6c1
Update show buffer
JinHai-CN Jun 6, 2024
ccbec5d
Update show buffer
JinHai-CN Jun 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ add_library(infinity_core
${expression_cpp}
${executor_cpp}
${common_cpp}
)
storage/buffer/file_worker/file_worker_type.cppm)

target_sources(infinity_core
PUBLIC
Expand Down
14 changes: 14 additions & 0 deletions src/executor/explain_physical_plan.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1242,6 +1242,20 @@ void ExplainPhysicalPlan::Explain(const PhysicalShow *show_node, SharedPtr<Vecto
result->emplace_back(MakeShared<String>(output_columns_str));
break;
}
case ShowType::kShowBuffer: {
String show_str;
if (intent_size != 0) {
show_str = String(intent_size - 2, ' ') + "-> SHOW BUFFER ";
} else {
show_str = "SHOW BUFFER ";
}
show_str += "(" + std::to_string(show_node->node_id()) + ")";
result->emplace_back(MakeShared<String>(show_str));

String output_columns_str = String(intent_size, ' ') + " - output columns: [path, status, size, buffered_type, type]";
result->emplace_back(MakeShared<String>(output_columns_str));
break;
}
case ShowType::kShowViews: {
String show_str;
if (intent_size != 0) {
Expand Down
106 changes: 105 additions & 1 deletion src/executor/operator/physical_show.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,8 @@ import chunk_index_entry;
import background_process;
import compaction_process;
import bg_task;
import buffer_obj;
import file_worker_type;

namespace infinity {

Expand Down Expand Up @@ -377,6 +379,21 @@ void PhysicalShow::Init() {
output_types_->emplace_back(varchar_type);
break;
}
case ShowType::kShowBuffer: {
output_names_->reserve(5);
output_types_->reserve(5);
output_names_->emplace_back("path");
output_names_->emplace_back("status");
output_names_->emplace_back("size");
output_names_->emplace_back("buffered_type");
output_names_->emplace_back("type");
output_types_->emplace_back(varchar_type);
output_types_->emplace_back(varchar_type);
output_types_->emplace_back(bigint_type);
output_types_->emplace_back(varchar_type);
output_types_->emplace_back(varchar_type);
break;
}
default: {
Status status = Status::NotSupport("Not implemented show type");
LOG_ERROR(status.message());
Expand Down Expand Up @@ -478,6 +495,10 @@ bool PhysicalShow::Execute(QueryContext *query_context, OperatorState *operator_
ExecuteShowConfig(query_context, show_operator_state);
break;
}
case ShowType::kShowBuffer: {
ExecuteShowBuffer(query_context, show_operator_state);
break;
}
default: {
String error_message = "Invalid chunk scan type";
LOG_CRITICAL(error_message);
Expand Down Expand Up @@ -2789,7 +2810,7 @@ void PhysicalShow::ExecuteShowIndexes(QueryContext *query_context, ShowOperatorS
{
auto map_guard = table_entry->IndexMetaMap();
for (const auto &[index_name, index_meta] : *map_guard) {
if (!output_block_ptr) {
if (output_block_ptr.get() == nullptr) {
output_block_ptr = DataBlock::MakeUniquePtr();
output_block_ptr->Init(column_types);
}
Expand Down Expand Up @@ -4080,4 +4101,87 @@ void PhysicalShow::ExecuteShowConfig(QueryContext *query_context, ShowOperatorSt
operator_state->output_.emplace_back(std::move(output_block_ptr));
}

void PhysicalShow::ExecuteShowBuffer(QueryContext *query_context, ShowOperatorState *operator_state) {
auto varchar_type = MakeShared<DataType>(LogicalType::kVarchar);
auto bigint_type = MakeShared<DataType>(LogicalType::kBigInt);

Vector<SharedPtr<ColumnDef>> column_defs = {
MakeShared<ColumnDef>(0, varchar_type, "path", std::set<ConstraintType>()),
MakeShared<ColumnDef>(1, varchar_type, "status", std::set<ConstraintType>()),
MakeShared<ColumnDef>(2, bigint_type, "size", std::set<ConstraintType>()),
MakeShared<ColumnDef>(3, varchar_type, "buffered_type", std::set<ConstraintType>()),
MakeShared<ColumnDef>(4, varchar_type, "type", std::set<ConstraintType>()),
};

SharedPtr<TableDef> table_def = TableDef::Make(MakeShared<String>("default_db"), MakeShared<String>("show_buffer"), column_defs);

// create data block for output state
Vector<SharedPtr<DataType>> column_types{
varchar_type,
varchar_type,
bigint_type,
varchar_type,
varchar_type,
};

UniquePtr<DataBlock> output_block_ptr = DataBlock::MakeUniquePtr();
output_block_ptr->Init(column_types);
SizeT row_count = 0;

BufferManager *buffer_manager = query_context->storage()->buffer_manager();
Vector<BufferObjectInfo> buffer_object_info_array = buffer_manager->GetBufferObjectsInfo();
for(const auto& buffer_object_info: buffer_object_info_array) {

if (output_block_ptr.get() == nullptr) {
output_block_ptr = DataBlock::MakeUniquePtr();
output_block_ptr->Init(column_types);
}

{
// path
Value value = Value::MakeVarchar(buffer_object_info.object_path_);
ValueExpression value_expr(value);
value_expr.AppendToChunk(output_block_ptr->column_vectors[0]);
}
{
// status
Value value = Value::MakeVarchar(BufferStatusToString(buffer_object_info.buffered_status_));
ValueExpression value_expr(value);
value_expr.AppendToChunk(output_block_ptr->column_vectors[1]);
}
{
// size
i64 buffer_object_size = static_cast<i64>(buffer_object_info.object_size_);
Value value = Value::MakeBigInt(buffer_object_size);
ValueExpression value_expr(value);
value_expr.AppendToChunk(output_block_ptr->column_vectors[2]);
}
{
// buffered type
Value value = Value::MakeVarchar(BufferTypeToString(buffer_object_info.buffered_type_));
ValueExpression value_expr(value);
value_expr.AppendToChunk(output_block_ptr->column_vectors[3]);
}
{
// type
Value value = Value::MakeVarchar(FileWorkerType2Str(buffer_object_info.file_type_));
ValueExpression value_expr(value);
value_expr.AppendToChunk(output_block_ptr->column_vectors[4]);
}

++ row_count;
if (row_count == output_block_ptr->capacity()) {
output_block_ptr->Finalize();
operator_state->output_.emplace_back(std::move(output_block_ptr));
output_block_ptr = nullptr;
row_count = 0;
}

}

output_block_ptr->Finalize();
operator_state->output_.emplace_back(std::move(output_block_ptr));
return ;
}

} // namespace infinity
2 changes: 2 additions & 0 deletions src/executor/operator/physical_show.cppm
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,8 @@ private:

void ExecuteShowConfig(QueryContext *query_context, ShowOperatorState *operator_state);

void ExecuteShowBuffer(QueryContext *query_context, ShowOperatorState *operator_state);

private:
ShowType scan_type_{ShowType::kInvalid};
String db_name_{};
Expand Down
Loading
Loading