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

refactor: proxy sql read #896

Merged
merged 2 commits into from
May 17, 2023
Merged

refactor: proxy sql read #896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8941df9
refactor: proxy sql read
chunshao90 May 16, 2023
3939968
refactor by CR
chunshao90 May 17, 2023
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
8 changes: 4 additions & 4 deletions integration_tests/cases/env/cluster/ddl/create_tables.result
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,11 +44,11 @@ affected_rows: 0

CREATE TABLE IF NOT EXISTS `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: CREATE TABLE IF NOT EXISTS `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t\", engine: \"Analytic\", create_if_not_exist: true, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: CREATE TABLE IF NOT EXISTS `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t\", engine: \"Analytic\", create_if_not_exist: true, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })

CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })

create table `05_create_tables_t2`(a int, b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic with (enable_ttl='false');

Expand All @@ -67,11 +67,11 @@ Int32(4),

create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t2\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t2\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })

create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t2\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to create table, msg:failed to create table by meta client, req:CreateTableRequest { schema_name: \"public\", name: \"05_create_tables_t2\", engine: \"Analytic\", create_if_not_exist: false, options: {}, partition_table_info: None }, err:Missing table info, msg:created table is not found in the create table response." })

create table `05_create_tables_t3`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Expand Down
6 changes: 3 additions & 3 deletions integration_tests/cases/env/cluster/ddl/partition_table.result
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ affected_rows: 0

DROP TABLE IF EXISTS `__partition_table_t_0`;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to create interpreter. Caused by: Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: DROP TABLE IF EXISTS `__partition_table_t_0`;. Caused by: Internal error, msg:Failed to create interpreter, err:Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })

CREATE TABLE `partition_table_t`(
`name`string TAG,
Expand All @@ -24,7 +24,7 @@ String("partition_table_t"),String("CREATE TABLE `partition_table_t` (`tsid` uin

SHOW CREATE TABLE __partition_table_t_0;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to create interpreter. Caused by: Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: SHOW CREATE TABLE __partition_table_t_0;. Caused by: Internal error, msg:Failed to create interpreter, err:Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })

INSERT INTO partition_table_t (t, name, value)
VALUES (1651737067000, "ceresdb0", 100),
Expand All @@ -43,7 +43,7 @@ affected_rows: 11

SELECT * from __partition_table_t_0 where name = "ceresdb0";

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to create interpreter. Caused by: Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: SELECT * from __partition_table_t_0 where name = \"ceresdb0\";. Caused by: Internal error, msg:Failed to create interpreter, err:Failed to check permission, msg:only can process sub tables in table partition directly when enable partition table access" })

SELECT * from partition_table_t where name = "ceresdb0";

Expand Down
6 changes: 3 additions & 3 deletions integration_tests/cases/env/local/ddl/create_tables.result
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ affected_rows: 0

CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"c1\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"c1\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: CREATE TABLE `05_create_tables_t`(c1 int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"c1\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"c1\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t." })

create table `05_create_tables_t2`(a int, b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic with (enable_ttl='false');

Expand All @@ -67,11 +67,11 @@ Int32(4),

create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t2\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"a\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"a\", default_value: None }, ColumnSchema { id: 4, name: \"b\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"b\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t2." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t2\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"a\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"a\", default_value: None }, ColumnSchema { id: 4, name: \"b\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"b\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t2." })

create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute interpreter, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t2\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"a\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"a\", default_value: None }, ColumnSchema { id: 4, name: \"b\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"b\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t2." })
Failed to execute query, err: Server(ServerError { code: 500, msg: "Failed to execute plan, sql: create table `05_create_tables_t2`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;. Caused by: Internal error, msg:Failed to execute interpreter, err:Failed to execute create table, err:Failed to create table by table manipulator, err:Failed to operate table, err:Failed to operate table, msg:failed to create table on shard, request:CreateTableRequest { catalog_name: \"ceresdb\", schema_name: \"public\", table_name: \"05_create_tables_t2\", table_id: None, table_schema: Schema { timestamp_index: 1, tsid_index: Some(0), column_schemas: ColumnSchemas { columns: [ColumnSchema { id: 1, name: \"tsid\", data_type: UInt64, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"tsid\", default_value: None }, ColumnSchema { id: 2, name: \"t\", data_type: Timestamp, is_nullable: false, is_tag: false, comment: \"\", escaped_name: \"t\", default_value: None }, ColumnSchema { id: 3, name: \"a\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"a\", default_value: None }, ColumnSchema { id: 4, name: \"b\", data_type: Int32, is_nullable: true, is_tag: false, comment: \"\", escaped_name: \"b\", default_value: None }] }, version: 1, primary_key_indexes: [0, 1] }, engine: \"Analytic\", options: {}, state: Stable, shard_id: 0, partition_info: None }, err:Failed to create table, table already exists, table:05_create_tables_t2." })

create table `05_create_tables_t3`(a int,b int, t timestamp NOT NULL, TIMESTAMP KEY(t)) ENGINE = Analytic;

Expand Down
40 changes: 5 additions & 35 deletions proxy/src/grpc/prom_query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ use common_types::{
};
use common_util::error::BoxError;
use http::StatusCode;
use interpreters::{context::Context as InterpreterContext, factory::Factory, interpreter::Output};
use interpreters::interpreter::Output;
use log::info;
use query_engine::executor::{Executor as QueryExecutor, RecordBatchVec};
use query_frontend::{
Expand Down Expand Up @@ -116,45 +116,15 @@ impl<Q: QueryExecutor + 'static> Proxy<Q> {
msg: "Query is blocked",
})?;

// Execute in interpreter
let interpreter_ctx = InterpreterContext::builder(request_id, deadline)
// Use current ctx's catalog and schema as default catalog and schema
.default_catalog_and_schema(catalog.to_string(), schema)
.build();
let interpreter_factory = Factory::new(
self.instance.query_executor.clone(),
self.instance.catalog_manager.clone(),
self.instance.table_engine.clone(),
self.instance.table_manipulator.clone(),
);
let interpreter = interpreter_factory
.create(interpreter_ctx, plan)
let output = self
.execute_plan(request_id, catalog, &schema, plan, deadline)
.await
.box_err()
.with_context(|| ErrWithCause {
code: StatusCode::INTERNAL_SERVER_ERROR,
msg: "Failed to create interpreter",
msg: "Failed to execute plan",
})?;

let output = if let Some(deadline) = deadline {
tokio::time::timeout_at(
tokio::time::Instant::from_std(deadline),
interpreter.execute(),
)
.await
.box_err()
.context(ErrWithCause {
code: StatusCode::REQUEST_TIMEOUT,
msg: "Query timeout",
})?
} else {
interpreter.execute().await
}
.box_err()
.context(ErrWithCause {
code: StatusCode::INTERNAL_SERVER_ERROR,
msg: "Failed to execute interpreter",
})?;

let resp = convert_output(output, column_name)
.box_err()
.context(ErrWithCause {
Expand Down
Loading