mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-09-10 07:22:17 +00:00
dennis zhuang
335a95a369
* feat(frontend): run entity-graph derivation as the caller The derivation contract requires the computed graph tables to run under the outer query's identity. Capture the caller's QueryContext when the computed table is resolved, thread it through EntityGraphProvider, and: - authorize every contributing source table against the caller via the new semantic_graph.query permission action, silently excluding denied sources (entities, edges and source_tables never appear); - execute the derivation plan under the caller's context so it inherits permissions, cancellation and deadline instead of a fresh default. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(operator): derive the entity-graph window from the scan's time predicate Implements the RFC window contract for the computed graph tables: - table: add extract_time_range_strict, a strict variant of the lenient time-range extraction that distinguishes an absent observed_at filter from one that cannot be safely turned into a range; - operator: replace GraphWindow with GraphQueryWindow, splitting the queried observed_at range from the source-scan range widened to whole 60s buckets, so boundary buckets aggregate over their full extent; - frontend: resolve the window from ScanRequest filters — no predicate keeps the last-hour default, a missing upper bound means now, and a missing lower bound or unextractable shape is an explicit error, never a silent fallback. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(operator): system-defined declared-edge table for the entity graph Reintroduces greptime_private.semantic_relationships_declared with a canonical, system-owned definition: - the CREATE TABLE expr (8-tag primary key, business validity columns, RED fields, 30d TTL); attributes is now a json column so the future union branch matches the computed table without a per-scan parse; - created on first use on every write path: SQL INSERT creates it before executing, and the gRPC row-insert auto-create substitutes the canonical expr instead of deriving a schema from the request; - user DDL (CREATE/ALTER/DROP/RENAME/TRUNCATE) and write-path auto-ALTER are rejected via the new is_ddl_reserved_table guard, while INSERT/DELETE stay allowed. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(operator): union declared edges into semantic_relationships Adds the declared-edge branch to the relationship derivation (build_relationships_plan replaces build_calls_plan): - latest revision per edge key first (mito dedups on primary key plus observed_at, so a re-asserted edge stores a new revision), then the business-validity overlap against the queried window; valid_from defaults to the declaration time and a NULL valid_until means the edge holds while its row exists; - the projected observed_at is synthesized inside the queried range (Inexact pushdown re-applies the scan's filters above the computed table, which would drop rows keyed by the physical revision time); window_end/fresh_until of open-ended edges take the window's upper bound so 'fresh_until >= now() - ...' queries see them; - tag columns are cast out of dictionary encoding, and the union is re-projected to the 16-column contract; - the frontend feeds the branch only when the physical table exists, the caller may read it, and its schema still matches the canonical definition (mismatch is an explicit error, not a silent drop). Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * test: cover declared edges, window contract and caller authorization - sqlness: system auto-create on first INSERT, latest-revision reads, open-ended vs retired validity, explicit/lower-only/upper-only window behavior, user-DDL rejection, rename-into rejection, DELETE cleanup; - integration: a permission checker denying one trace table excludes it from both semantic_relationships and semantic_entities. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: allow DROP/TRUNCATE on the declared-edge table and fix CI lints The definition guard rejected every DDL, which left sqlness (and any shared deployment) no way to remove the table the semantic_graph case creates — its extra region then broke unrelated region/partition case expectations. Narrow the guard to what actually protects the canonical definition: user CREATE, ALTER, RENAME-into and repartition stay rejected, while DROP and TRUNCATE are allowed — dropping loses nothing structural, the next INSERT recreates the table canonically, and DROP doubles as the recovery path if the canonical definition ever changes. The sqlness case now verifies drop-then-recreate and cleans up after itself. Also: rustfmt for the catalog crate and two typo fixes. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * chore: adapt canonical declared-table create to TriggerReason Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: address review on the declared-edge table lifecycle and revision reads - gRPC first writes actually work now: the reserved table's creation went through the generic create_table_inner, which the definition guard itself rejects; both branches of create_or_alter_tables_on_demand route it to create_declared_relationships_table instead, and being a system action it also bypasses the auto_create_table config/hint; - revision selection is as-of the queried window: revisions recorded after the window's end, or whose validity starts after it, no longer outrank (and hide) the revision that was in effect inside it; - the canonical-schema check validates the whole definition the union semantics lean on — time index, primary key, engine, append/merge mode — not just column names and types; - UNDROP TABLE of the reserved name is rejected like CREATE: it could resurrect a pre-canonical shape, and the next INSERT recreates the table anyway. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * chore: trim over-commenting in the entity-graph code Comments that restated adjacent code or narrated justification are cut; the ones stating non-obvious contracts and gotchas stay. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: reject CREATE VIEW against DDL-reserved table names A view named greptime_private.semantic_relationships_declared would squat the reserved name: the first INSERT then skips the canonical create (an object already exists) and graph reads fail on the schema mismatch. CREATE VIEW now passes the same definition guard as CREATE TABLE. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat: debug-log authorization exclusions; declared-edge TTL to 90d Sources the derivation contract silently excludes (per-table denial, whole-scan denial, the declared-edge table) are invisible from outside; a debug log at each names what was excluded and why. The declared-edge table's default TTL becomes 90d, overridable at creation time via GREPTIMEDB_DECLARED_RELATIONSHIPS_TTL (a proper configuration option is a TODO). Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: rank declared-edge revisions by the visible edge identity Ranking partitioned by the full primary key, but the projection drops scope and generation_id: two assertions of the same visible edge under different generations both ranked first and came out as duplicate, indistinguishable rows. Rank by the exposed identity (endpoints, rel_type, provenance) instead, with generation_id/scope as deterministic tie-breakers for same-timestamp assertions. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * test: drop redundant declared-edge tests The generations regression is already asserted by the revision and as-of tests; the DDL shape test restated the declarative builder against itself. Its one non-tautological check (attributes maps to the json type) moves into the schema-matcher test. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix: reject disjunctive graph windows and unmatchable future windows - OR/IN over observed_at collapse disjoint ranges into their convex hull; a declared edge's synthesized timestamp can land in a gap and be dropped by the re-applied filter even though the edge is valid at a requested instant. The strict extractor now rejects those shapes. - A lower bound in the future inverts against the implicit up-to-now upper bound; the declared branch then fabricated an edge observed at the future bound. Such windows now derive nothing. - The reserved-table gRPC create path classifies an instant-TTL table like every sibling path. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
1516 lines
65 KiB
Rust
1516 lines
65 KiB
Rust
// Copyright 2023 Greptime Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
mod flight;
|
|
mod network;
|
|
|
|
use api::v1::QueryRequest;
|
|
use api::v1::greptime_request::Request;
|
|
use api::v1::query_request::Query;
|
|
use common_query::OutputData;
|
|
use common_recordbatch::RecordBatches;
|
|
use frontend::instance::Instance;
|
|
use servers::query_handler::grpc::GrpcQueryHandler;
|
|
use session::context::QueryContext;
|
|
|
|
#[allow(unused)]
|
|
async fn query_and_expect(instance: &Instance, sql: &str, expected: &str) {
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql(sql.to_string())),
|
|
});
|
|
let output = GrpcQueryHandler::do_query(instance, request, QueryContext::arc())
|
|
.await
|
|
.unwrap();
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let actual = recordbatches.pretty_print().unwrap();
|
|
assert_eq!(actual, expected, "actual: {}", actual);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod test {
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
|
|
use api::v1::column::Values;
|
|
use api::v1::column_data_type_extension::TypeExt;
|
|
use api::v1::ddl_request::Expr as DdlExpr;
|
|
use api::v1::greptime_request::Request;
|
|
use api::v1::query_request::Query;
|
|
use api::v1::region::QueryRequest as RegionQueryRequest;
|
|
use api::v1::{
|
|
AddColumn, AddColumns, AlterTableExpr, Column, ColumnDataType, ColumnDataTypeExtension,
|
|
ColumnDef, CreateDatabaseExpr, CreateTableExpr, DdlRequest, DeleteRequest, DeleteRequests,
|
|
DropTableExpr, InsertIntoPlan, InsertRequest, InsertRequests, QueryRequest, SemanticType,
|
|
VectorTypeExtension, alter_table_expr,
|
|
};
|
|
use auth::{
|
|
DefaultPermissionChecker, Identity, Password, PermissionCheckerRef, UserProvider,
|
|
static_user_provider_from_option,
|
|
};
|
|
use client::OutputData;
|
|
use common_base::Plugins;
|
|
use common_catalog::consts::{
|
|
DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE,
|
|
SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME,
|
|
};
|
|
use common_meta::rpc::router::region_distribution;
|
|
use common_query::Output;
|
|
use common_query::logical_plan::breakup_insert_plan;
|
|
use common_recordbatch::RecordBatches;
|
|
use frontend::instance::Instance;
|
|
use query::parser::QueryLanguageParser;
|
|
use query::query_engine::DefaultSerializer;
|
|
use rstest::rstest;
|
|
use rstest_reuse::apply;
|
|
use servers::query_handler::grpc::GrpcQueryHandler;
|
|
use session::context::{QueryContext, QueryContextBuilder};
|
|
use store_api::mito_engine_options::TWCS_TIME_WINDOW;
|
|
use store_api::storage::RegionId;
|
|
use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
|
|
|
|
use super::*;
|
|
use crate::standalone::GreptimeDbStandaloneBuilder;
|
|
use crate::test_util::execute_sql_and_expect;
|
|
use crate::tests;
|
|
use crate::tests::MockDistributedInstance;
|
|
use crate::tests::test_util::{MockInstance, both_instances_cases, distributed, standalone};
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_distributed_handle_ddl_request() {
|
|
common_telemetry::init_default_ut_logging();
|
|
let instance =
|
|
tests::create_distributed_instance("test_distributed_handle_ddl_request").await;
|
|
|
|
test_handle_ddl_request(instance.frontend().as_ref()).await;
|
|
|
|
verify_table_is_dropped(&instance).await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_standalone_handle_ddl_request() {
|
|
let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_handle_ddl_request")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
test_handle_ddl_request(instance.as_ref()).await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_distributed_handle_multi_ddl_request() {
|
|
common_telemetry::init_default_ut_logging();
|
|
let instance =
|
|
tests::create_distributed_instance("test_distributed_handle_multi_ddl_request").await;
|
|
|
|
test_handle_multi_ddl_request(instance.frontend().as_ref()).await;
|
|
|
|
verify_table_is_dropped(&instance).await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_standalone_handle_multi_ddl_request() {
|
|
let standalone =
|
|
GreptimeDbStandaloneBuilder::new("test_standalone_handle_multi_ddl_request")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
test_handle_multi_ddl_request(instance.as_ref()).await;
|
|
}
|
|
|
|
async fn query(instance: &Instance, request: Request) -> Output {
|
|
GrpcQueryHandler::do_query(instance, request, QueryContext::arc())
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_grpc_insert_into_plan_rejects_readonly_user() {
|
|
let plugins = Plugins::new();
|
|
plugins.insert::<PermissionCheckerRef>(DefaultPermissionChecker::arc());
|
|
|
|
let standalone =
|
|
GreptimeDbStandaloneBuilder::new("test_grpc_insert_into_plan_rejects_readonly_user")
|
|
.with_plugin(plugins)
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
let table_name = "grpc_insert_into_plan_auth";
|
|
|
|
create_table(
|
|
instance,
|
|
format!("CREATE TABLE {table_name} (host STRING, val DOUBLE, ts TIMESTAMP TIME INDEX)"),
|
|
)
|
|
.await;
|
|
|
|
let stmt = QueryLanguageParser::parse_sql(
|
|
&format!("INSERT INTO {table_name} VALUES ('readonly-bypass', 42.0, 1000)"),
|
|
&QueryContext::arc(),
|
|
)
|
|
.unwrap();
|
|
let plan = instance
|
|
.statement_executor()
|
|
.plan(&stmt, QueryContext::arc())
|
|
.await
|
|
.unwrap();
|
|
let (table_name, insert_plan) = breakup_insert_plan(&plan, "greptime", "public").unwrap();
|
|
let logical_plan = DFLogicalSubstraitConvertor
|
|
.encode(&insert_plan, DefaultSerializer)
|
|
.unwrap()
|
|
.to_vec();
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::InsertIntoPlan(InsertIntoPlan {
|
|
table_name: Some(table_name),
|
|
logical_plan,
|
|
})),
|
|
});
|
|
let ctx = QueryContext::arc();
|
|
let provider =
|
|
static_user_provider_from_option("static_user_provider:cmd:readonly:ro=readonly_pwd")
|
|
.unwrap();
|
|
let readonly_user = provider
|
|
.authenticate(
|
|
Identity::UserId("readonly", None),
|
|
Password::PlainText("readonly_pwd".to_string().into()),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
ctx.set_current_user(readonly_user);
|
|
|
|
let err = GrpcQueryHandler::do_query(instance.as_ref(), request, ctx)
|
|
.await
|
|
.unwrap_err();
|
|
let err_msg = format!("{err:?}");
|
|
assert!(
|
|
err_msg.contains("not authorized"),
|
|
"unexpected error: {err_msg}"
|
|
);
|
|
|
|
query_and_expect(
|
|
instance,
|
|
"SELECT count(*) FROM grpc_insert_into_plan_auth",
|
|
"\
|
|
+----------+
|
|
| count(*) |
|
|
+----------+
|
|
| 0 |
|
|
+----------+",
|
|
)
|
|
.await;
|
|
}
|
|
|
|
async fn test_handle_multi_ddl_request(instance: &Instance) {
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::CreateDatabase(CreateDatabaseExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
create_if_not_exists: true,
|
|
options: Default::default(),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::CreateTable(CreateTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
column_defs: vec![
|
|
ColumnDef {
|
|
name: "a".to_string(),
|
|
data_type: ColumnDataType::String as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
},
|
|
ColumnDef {
|
|
name: "ts".to_string(),
|
|
data_type: ColumnDataType::TimestampMillisecond as _,
|
|
is_nullable: false,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
time_index: "ts".to_string(),
|
|
engine: MITO_ENGINE.to_string(),
|
|
..Default::default()
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::AlterTable(AlterTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
kind: Some(alter_table_expr::Kind::AddColumns(AddColumns {
|
|
add_columns: vec![
|
|
AddColumn {
|
|
column_def: Some(ColumnDef {
|
|
name: "b".to_string(),
|
|
data_type: ColumnDataType::Int32 as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
}),
|
|
location: None,
|
|
add_if_not_exists: true,
|
|
},
|
|
AddColumn {
|
|
column_def: Some(ColumnDef {
|
|
name: "a".to_string(),
|
|
data_type: ColumnDataType::String as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
}),
|
|
location: None,
|
|
add_if_not_exists: true,
|
|
},
|
|
],
|
|
})),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::AlterTable(AlterTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
kind: Some(alter_table_expr::Kind::AddColumns(AddColumns {
|
|
add_columns: vec![
|
|
AddColumn {
|
|
column_def: Some(ColumnDef {
|
|
name: "c".to_string(),
|
|
data_type: ColumnDataType::Int32 as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
}),
|
|
location: None,
|
|
add_if_not_exists: true,
|
|
},
|
|
AddColumn {
|
|
column_def: Some(ColumnDef {
|
|
name: "d".to_string(),
|
|
data_type: ColumnDataType::Int32 as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
}),
|
|
location: None,
|
|
add_if_not_exists: true,
|
|
},
|
|
],
|
|
})),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql("INSERT INTO database_created_through_grpc.table_created_through_grpc (a, b, c, d, ts) VALUES ('s', 1, 1, 1, 1672816466000)".to_string()))
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
|
|
let sql = "SELECT ts, a, b FROM database_created_through_grpc.table_created_through_grpc";
|
|
let expected = "\
|
|
+---------------------+---+---+
|
|
| ts | a | b |
|
|
+---------------------+---+---+
|
|
| 2023-01-04T07:14:26 | s | 1 |
|
|
+---------------------+---+---+";
|
|
query_and_expect(instance, sql, expected).await;
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::DropTable(DropTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
..Default::default()
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
}
|
|
|
|
async fn test_handle_ddl_request(instance: &Instance) {
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::CreateDatabase(CreateDatabaseExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
create_if_not_exists: true,
|
|
options: Default::default(),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::CreateTable(CreateTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
column_defs: vec![
|
|
ColumnDef {
|
|
name: "a".to_string(),
|
|
data_type: ColumnDataType::String as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
},
|
|
ColumnDef {
|
|
name: "ts".to_string(),
|
|
data_type: ColumnDataType::TimestampMillisecond as _,
|
|
is_nullable: false,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
time_index: "ts".to_string(),
|
|
engine: MITO_ENGINE.to_string(),
|
|
..Default::default()
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::AlterTable(AlterTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
kind: Some(alter_table_expr::Kind::AddColumns(AddColumns {
|
|
add_columns: vec![AddColumn {
|
|
column_def: Some(ColumnDef {
|
|
name: "b".to_string(),
|
|
data_type: ColumnDataType::Int32 as _,
|
|
is_nullable: true,
|
|
default_constraint: vec![],
|
|
semantic_type: SemanticType::Field as i32,
|
|
..Default::default()
|
|
}),
|
|
location: None,
|
|
add_if_not_exists: false,
|
|
}],
|
|
})),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql("INSERT INTO database_created_through_grpc.table_created_through_grpc (a, b, ts) VALUES ('s', 1, 1672816466000)".to_string()))
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
|
|
let sql = "SELECT ts, a, b FROM database_created_through_grpc.table_created_through_grpc";
|
|
let expected = "\
|
|
+---------------------+---+---+
|
|
| ts | a | b |
|
|
+---------------------+---+---+
|
|
| 2023-01-04T07:14:26 | s | 1 |
|
|
+---------------------+---+---+";
|
|
query_and_expect(instance, sql, expected).await;
|
|
|
|
let request = Request::Ddl(DdlRequest {
|
|
expr: Some(DdlExpr::DropTable(DropTableExpr {
|
|
catalog_name: "greptime".to_string(),
|
|
schema_name: "database_created_through_grpc".to_string(),
|
|
table_name: "table_created_through_grpc".to_string(),
|
|
..Default::default()
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
}
|
|
|
|
async fn verify_table_is_dropped(instance: &MockDistributedInstance) {
|
|
assert!(
|
|
instance
|
|
.frontend()
|
|
.catalog_manager()
|
|
.table(
|
|
"greptime",
|
|
"database_created_through_grpc",
|
|
"table_created_through_grpc",
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_distributed_insert_delete_and_query() {
|
|
common_telemetry::init_default_ut_logging();
|
|
|
|
let instance =
|
|
tests::create_distributed_instance("test_distributed_insert_delete_and_query").await;
|
|
let frontend = instance.frontend();
|
|
let frontend = frontend.as_ref();
|
|
|
|
let table_name = "my_dist_table";
|
|
let sql = format!(
|
|
r"
|
|
CREATE TABLE {table_name} (
|
|
a INT,
|
|
b STRING,
|
|
c JSON,
|
|
d VECTOR(3),
|
|
ts TIMESTAMP,
|
|
TIME INDEX (ts),
|
|
PRIMARY KEY (a, b, c)
|
|
) PARTITION ON COLUMNS(a) (
|
|
a < 10,
|
|
a >= 10 AND a < 20,
|
|
a >= 20 AND a < 50,
|
|
a >= 50
|
|
)"
|
|
);
|
|
create_table(frontend, sql).await;
|
|
|
|
test_insert_delete_and_query_on_existing_table(frontend, table_name).await;
|
|
|
|
verify_data_distribution(
|
|
&instance,
|
|
table_name,
|
|
HashMap::from([
|
|
(
|
|
0u32,
|
|
"\
|
|
+---------------------+---+-------------------+
|
|
| ts | a | b |
|
|
+---------------------+---+-------------------+
|
|
| 2023-01-01T07:26:12 | 1 | ts: 1672557972000 |
|
|
| 2023-01-01T07:26:15 | 4 | ts: 1672557975000 |
|
|
| 2023-01-01T07:26:16 | 5 | ts: 1672557976000 |
|
|
| 2023-01-01T07:26:17 | | ts: 1672557977000 |
|
|
+---------------------+---+-------------------+",
|
|
),
|
|
(
|
|
1u32,
|
|
"\
|
|
+---------------------+----+-------------------+
|
|
| ts | a | b |
|
|
+---------------------+----+-------------------+
|
|
| 2023-01-01T07:26:18 | 11 | ts: 1672557978000 |
|
|
+---------------------+----+-------------------+",
|
|
),
|
|
(
|
|
2u32,
|
|
"\
|
|
+---------------------+----+-------------------+
|
|
| ts | a | b |
|
|
+---------------------+----+-------------------+
|
|
| 2023-01-01T07:26:20 | 20 | ts: 1672557980000 |
|
|
| 2023-01-01T07:26:21 | 21 | ts: 1672557981000 |
|
|
| 2023-01-01T07:26:23 | 23 | ts: 1672557983000 |
|
|
+---------------------+----+-------------------+",
|
|
),
|
|
(
|
|
3u32,
|
|
"\
|
|
+---------------------+----+-------------------+
|
|
| ts | a | b |
|
|
+---------------------+----+-------------------+
|
|
| 2023-01-01T07:26:24 | 50 | ts: 1672557984000 |
|
|
| 2023-01-01T07:26:25 | 51 | ts: 1672557985000 |
|
|
+---------------------+----+-------------------+",
|
|
),
|
|
]),
|
|
)
|
|
.await;
|
|
|
|
test_insert_delete_and_query_on_auto_created_table(frontend).await;
|
|
|
|
// Auto created table has only one region.
|
|
verify_data_distribution(
|
|
&instance,
|
|
"auto_created_table",
|
|
HashMap::from([(
|
|
0u32,
|
|
"\
|
|
+---------------------+---+---+
|
|
| ts | a | b |
|
|
+---------------------+---+---+
|
|
| 2023-01-01T07:26:16 | | |
|
|
| 2023-01-01T07:26:17 | 6 | |
|
|
| 2023-01-01T07:26:18 | | x |
|
|
| 2023-01-01T07:26:20 | | z |
|
|
+---------------------+---+---+",
|
|
)]),
|
|
)
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_standalone_insert_and_query() {
|
|
common_telemetry::init_default_ut_logging();
|
|
let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_insert_and_query")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
let table_name = "my_table";
|
|
let sql = format!(
|
|
"CREATE TABLE {table_name} (a INT, b STRING, c JSON, ts TIMESTAMP, TIME INDEX (ts), PRIMARY KEY (a, b, c))"
|
|
);
|
|
create_table(instance, sql).await;
|
|
|
|
test_insert_delete_and_query_on_existing_table(instance, table_name).await;
|
|
|
|
test_insert_delete_and_query_on_auto_created_table(instance).await
|
|
}
|
|
|
|
async fn create_table(frontend: &Instance, sql: String) {
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql(sql)),
|
|
});
|
|
let output = query(frontend, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(0)));
|
|
}
|
|
|
|
async fn test_insert_delete_and_query_on_existing_table(instance: &Instance, table_name: &str) {
|
|
let timestamp_millisecond_values = vec![
|
|
1672557972000,
|
|
1672557973000,
|
|
1672557974000,
|
|
1672557975000,
|
|
1672557976000,
|
|
1672557977000,
|
|
1672557978000,
|
|
1672557979000,
|
|
1672557980000,
|
|
1672557981000,
|
|
1672557982000,
|
|
1672557983000,
|
|
1672557984000,
|
|
1672557985000,
|
|
1672557986000,
|
|
1672557987000,
|
|
];
|
|
let json_strings = vec![
|
|
r#"{ "id": 1, "name": "Alice", "age": 30, "active": true }"#.to_string(),
|
|
r#"{ "id": 2, "name": "Bob", "balance": 1234.56, "active": false }"#.to_string(),
|
|
r#"{ "id": 3, "tags": ["rust", "testing", "json"], "age": 28 }"#.to_string(),
|
|
r#"{ "id": 4, "metadata": { "created_at": "2024-10-30T12:00:00Z", "status": "inactive" } }"#.to_string(),
|
|
r#"{ "id": 5, "name": null, "phone": "+1234567890" }"#.to_string(),
|
|
r#"{ "id": 6, "height": 5.9, "weight": 72.5, "active": true }"#.to_string(),
|
|
r#"{ "id": 7, "languages": ["English", "Spanish"], "age": 29 }"#.to_string(),
|
|
r#"{ "id": 8, "contact": { "email": "hank@example.com", "phone": "+0987654321" } }"#.to_string(),
|
|
r#"{ "id": 9, "preferences": { "notifications": true, "theme": "dark" } }"#.to_string(),
|
|
r#"{ "id": 10, "scores": [88, 92, 76], "active": false }"#.to_string(),
|
|
r#"{ "id": 11, "birthday": "1996-07-20", "location": { "city": "New York", "zip": "10001" } }"#.to_string(),
|
|
r#"{ "id": 12, "subscription": { "type": "premium", "expires": "2025-01-01" } }"#.to_string(),
|
|
r#"{ "id": 13, "settings": { "volume": 0.8, "brightness": 0.6 }, "active": true }"#.to_string(),
|
|
r#"{ "id": 14, "notes": ["first note", "second note"], "priority": 1 }"#.to_string(),
|
|
r#"{ "id": 15, "transactions": [{ "amount": 500, "date": "2024-01-01" }, { "amount": -200, "date": "2024-02-01" }] }"#.to_string(),
|
|
r#"{ "id": 16, "transactions": [{ "amount": 500, "date": "2024-01-01" }] }"#.to_string(),
|
|
];
|
|
let vector_values = [
|
|
[1.0f32, 2.0, 3.0],
|
|
[4.0, 5.0, 6.0],
|
|
[7.0, 8.0, 9.0],
|
|
[10.0, 11.0, 12.0],
|
|
[13.0, 14.0, 15.0],
|
|
[16.0, 17.0, 18.0],
|
|
[19.0, 20.0, 21.0],
|
|
[22.0, 23.0, 24.0],
|
|
[25.0, 26.0, 27.0],
|
|
[28.0, 29.0, 30.0],
|
|
[31.0, 32.0, 33.0],
|
|
[34.0, 35.0, 36.0],
|
|
[37.0, 38.0, 39.0],
|
|
[40.0, 41.0, 42.0],
|
|
[43.0, 44.0, 45.0],
|
|
[46.0, 47.0, 48.0],
|
|
]
|
|
.iter()
|
|
.map(|x| x.iter().flat_map(|&f| f.to_le_bytes()).collect::<Vec<u8>>())
|
|
.collect::<Vec<_>>();
|
|
|
|
let insert = InsertRequest {
|
|
table_name: table_name.to_string(),
|
|
columns: vec![
|
|
Column {
|
|
column_name: "a".to_string(),
|
|
values: Some(Values {
|
|
i32_values: vec![1, 2, 3, 4, 5, 11, 12, 20, 21, 22, 23, 50, 51, 52, 53],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![32, 0],
|
|
semantic_type: SemanticType::Tag as i32,
|
|
datatype: ColumnDataType::Int32 as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "b".to_string(),
|
|
values: Some(Values {
|
|
string_values: timestamp_millisecond_values
|
|
.iter()
|
|
.map(|x| format!("ts: {x}"))
|
|
.collect(),
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
datatype: ColumnDataType::String as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "c".to_string(),
|
|
values: Some(Values {
|
|
string_values: json_strings,
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
datatype: ColumnDataType::Json as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "d".to_string(),
|
|
values: Some(Values {
|
|
binary_values: vector_values.clone(),
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Vector as i32,
|
|
datatype_extension: Some(ColumnDataTypeExtension {
|
|
type_ext: Some(TypeExt::VectorType(VectorTypeExtension { dim: 3 })),
|
|
}),
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values,
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count: 16,
|
|
};
|
|
let output = query(
|
|
instance,
|
|
Request::Inserts(InsertRequests {
|
|
inserts: vec![insert],
|
|
}),
|
|
)
|
|
.await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(16)));
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql(format!(
|
|
"SELECT ts, a, b, json_to_string(c) as c, d FROM {table_name} ORDER BY ts"
|
|
))),
|
|
});
|
|
let output = query(instance, request.clone()).await;
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let expected = r#"+---------------------+----+-------------------+---------------------------------------------------------------------------------------------------+--------------------------+
|
|
| ts | a | b | c | d |
|
|
+---------------------+----+-------------------+---------------------------------------------------------------------------------------------------+--------------------------+
|
|
| 2023-01-01T07:26:12 | 1 | ts: 1672557972000 | {"active":true,"age":30,"id":1,"name":"Alice"} | 0000803f0000004000004040 |
|
|
| 2023-01-01T07:26:13 | 2 | ts: 1672557973000 | {"active":false,"balance":1234.56,"id":2,"name":"Bob"} | 000080400000a0400000c040 |
|
|
| 2023-01-01T07:26:14 | 3 | ts: 1672557974000 | {"age":28,"id":3,"tags":["rust","testing","json"]} | 0000e0400000004100001041 |
|
|
| 2023-01-01T07:26:15 | 4 | ts: 1672557975000 | {"id":4,"metadata":{"created_at":"2024-10-30T12:00:00Z","status":"inactive"}} | 000020410000304100004041 |
|
|
| 2023-01-01T07:26:16 | 5 | ts: 1672557976000 | {"id":5,"name":null,"phone":"+1234567890"} | 000050410000604100007041 |
|
|
| 2023-01-01T07:26:17 | | ts: 1672557977000 | {"active":true,"height":5.9,"id":6,"weight":72.5} | 000080410000884100009041 |
|
|
| 2023-01-01T07:26:18 | 11 | ts: 1672557978000 | {"age":29,"id":7,"languages":["English","Spanish"]} | 000098410000a0410000a841 |
|
|
| 2023-01-01T07:26:19 | 12 | ts: 1672557979000 | {"contact":{"email":"hank@example.com","phone":"+0987654321"},"id":8} | 0000b0410000b8410000c041 |
|
|
| 2023-01-01T07:26:20 | 20 | ts: 1672557980000 | {"id":9,"preferences":{"notifications":true,"theme":"dark"}} | 0000c8410000d0410000d841 |
|
|
| 2023-01-01T07:26:21 | 21 | ts: 1672557981000 | {"active":false,"id":10,"scores":[88,92,76]} | 0000e0410000e8410000f041 |
|
|
| 2023-01-01T07:26:22 | 22 | ts: 1672557982000 | {"birthday":"1996-07-20","id":11,"location":{"city":"New York","zip":"10001"}} | 0000f8410000004200000442 |
|
|
| 2023-01-01T07:26:23 | 23 | ts: 1672557983000 | {"id":12,"subscription":{"expires":"2025-01-01","type":"premium"}} | 0000084200000c4200001042 |
|
|
| 2023-01-01T07:26:24 | 50 | ts: 1672557984000 | {"active":true,"id":13,"settings":{"brightness":0.6,"volume":0.8}} | 000014420000184200001c42 |
|
|
| 2023-01-01T07:26:25 | 51 | ts: 1672557985000 | {"id":14,"notes":["first note","second note"],"priority":1} | 000020420000244200002842 |
|
|
| 2023-01-01T07:26:26 | 52 | ts: 1672557986000 | {"id":15,"transactions":[{"amount":500,"date":"2024-01-01"},{"amount":-200,"date":"2024-02-01"}]} | 00002c420000304200003442 |
|
|
| 2023-01-01T07:26:27 | 53 | ts: 1672557987000 | {"id":16,"transactions":[{"amount":500,"date":"2024-01-01"}]} | 0000384200003c4200004042 |
|
|
+---------------------+----+-------------------+---------------------------------------------------------------------------------------------------+--------------------------+"#;
|
|
similar_asserts::assert_eq!(recordbatches.pretty_print().unwrap(), expected);
|
|
|
|
// Checks if the encoded vector values are as expected.
|
|
let hex_repr_of_vector_values = vector_values.iter().map(hex::encode).collect::<Vec<_>>();
|
|
assert_eq!(
|
|
hex_repr_of_vector_values,
|
|
vec![
|
|
"0000803f0000004000004040",
|
|
"000080400000a0400000c040",
|
|
"0000e0400000004100001041",
|
|
"000020410000304100004041",
|
|
"000050410000604100007041",
|
|
"000080410000884100009041",
|
|
"000098410000a0410000a841",
|
|
"0000b0410000b8410000c041",
|
|
"0000c8410000d0410000d841",
|
|
"0000e0410000e8410000f041",
|
|
"0000f8410000004200000442",
|
|
"0000084200000c4200001042",
|
|
"000014420000184200001c42",
|
|
"000020420000244200002842",
|
|
"00002c420000304200003442",
|
|
"0000384200003c4200004042",
|
|
]
|
|
);
|
|
|
|
let new_grpc_delete_request = |a, b, c, d, ts, row_count| DeleteRequest {
|
|
table_name: table_name.to_string(),
|
|
key_columns: vec![
|
|
Column {
|
|
column_name: "a".to_string(),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
values: Some(Values {
|
|
i32_values: a,
|
|
..Default::default()
|
|
}),
|
|
datatype: ColumnDataType::Int32 as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "b".to_string(),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
values: Some(Values {
|
|
string_values: b,
|
|
..Default::default()
|
|
}),
|
|
datatype: ColumnDataType::String as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "c".to_string(),
|
|
values: Some(Values {
|
|
string_values: c,
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
datatype: ColumnDataType::Json as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "d".to_string(),
|
|
values: Some(Values {
|
|
binary_values: d,
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Vector as i32,
|
|
datatype_extension: Some(ColumnDataTypeExtension {
|
|
type_ext: Some(TypeExt::VectorType(VectorTypeExtension { dim: 3 })),
|
|
}),
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: ts,
|
|
..Default::default()
|
|
}),
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count,
|
|
};
|
|
let delete1 = new_grpc_delete_request(
|
|
vec![2, 12, 22, 52],
|
|
vec![
|
|
"ts: 1672557973000".to_string(),
|
|
"ts: 1672557979000".to_string(),
|
|
"ts: 1672557982000".to_string(),
|
|
"ts: 1672557986000".to_string(),
|
|
],
|
|
vec![
|
|
r#"{ "id": 2, "name": "Bob", "balance": 1234.56, "active": false }"#.to_string(),
|
|
r#"{ "id": 8, "contact": { "email": "hank@example.com", "phone": "+0987654321" } }"#.to_string(),
|
|
r#"{ "id": 11, "birthday": "1996-07-20", "location": { "city": "New York", "zip": "10001" } }"#.to_string(),
|
|
r#"{ "id": 15, "transactions": [{ "amount": 500, "date": "2024-01-01" }, { "amount": -200, "date": "2024-02-01" }] }"#.to_string(),
|
|
],
|
|
vec![
|
|
[4.0f32, 5.0, 6.0].iter().flat_map(|f| f.to_le_bytes()).collect::<Vec<u8>>(),
|
|
[22.0f32, 23.0, 24.0].iter().flat_map(|f| f.to_le_bytes()).collect::<Vec<u8>>(),
|
|
[31.0f32, 32.0, 33.0].iter().flat_map(|f| f.to_le_bytes()).collect::<Vec<u8>>(),
|
|
[43.0f32, 44.0, 45.0].iter().flat_map(|f| f.to_le_bytes()).collect::<Vec<u8>>(),
|
|
],
|
|
vec![1672557973000, 1672557979000, 1672557982000, 1672557986000],
|
|
4,
|
|
);
|
|
let delete2 = new_grpc_delete_request(
|
|
vec![3, 53],
|
|
vec![
|
|
"ts: 1672557974000".to_string(),
|
|
"ts: 1672557987000".to_string(),
|
|
],
|
|
vec![
|
|
r#"{ "id": 3, "tags": ["rust", "testing", "json"], "age": 28 }"#.to_string(),
|
|
r#"{ "id": 16, "transactions": [{ "amount": 500, "date": "2024-01-01" }] }"#
|
|
.to_string(),
|
|
],
|
|
vec![
|
|
[7.0f32, 8.0, 9.0]
|
|
.iter()
|
|
.flat_map(|f| f.to_le_bytes())
|
|
.collect::<Vec<u8>>(),
|
|
[46.0f32, 47.0, 48.0]
|
|
.iter()
|
|
.flat_map(|f| f.to_le_bytes())
|
|
.collect::<Vec<u8>>(),
|
|
],
|
|
vec![1672557974000, 1672557987000],
|
|
2,
|
|
);
|
|
let output = query(
|
|
instance,
|
|
Request::Deletes(DeleteRequests {
|
|
deletes: vec![delete1, delete2],
|
|
}),
|
|
)
|
|
.await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(6)));
|
|
|
|
let output = query(instance, request).await;
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let expected = r#"+---------------------+----+-------------------+-------------------------------------------------------------------------------+--------------------------+
|
|
| ts | a | b | c | d |
|
|
+---------------------+----+-------------------+-------------------------------------------------------------------------------+--------------------------+
|
|
| 2023-01-01T07:26:12 | 1 | ts: 1672557972000 | {"active":true,"age":30,"id":1,"name":"Alice"} | 0000803f0000004000004040 |
|
|
| 2023-01-01T07:26:15 | 4 | ts: 1672557975000 | {"id":4,"metadata":{"created_at":"2024-10-30T12:00:00Z","status":"inactive"}} | 000020410000304100004041 |
|
|
| 2023-01-01T07:26:16 | 5 | ts: 1672557976000 | {"id":5,"name":null,"phone":"+1234567890"} | 000050410000604100007041 |
|
|
| 2023-01-01T07:26:17 | | ts: 1672557977000 | {"active":true,"height":5.9,"id":6,"weight":72.5} | 000080410000884100009041 |
|
|
| 2023-01-01T07:26:18 | 11 | ts: 1672557978000 | {"age":29,"id":7,"languages":["English","Spanish"]} | 000098410000a0410000a841 |
|
|
| 2023-01-01T07:26:20 | 20 | ts: 1672557980000 | {"id":9,"preferences":{"notifications":true,"theme":"dark"}} | 0000c8410000d0410000d841 |
|
|
| 2023-01-01T07:26:21 | 21 | ts: 1672557981000 | {"active":false,"id":10,"scores":[88,92,76]} | 0000e0410000e8410000f041 |
|
|
| 2023-01-01T07:26:23 | 23 | ts: 1672557983000 | {"id":12,"subscription":{"expires":"2025-01-01","type":"premium"}} | 0000084200000c4200001042 |
|
|
| 2023-01-01T07:26:24 | 50 | ts: 1672557984000 | {"active":true,"id":13,"settings":{"brightness":0.6,"volume":0.8}} | 000014420000184200001c42 |
|
|
| 2023-01-01T07:26:25 | 51 | ts: 1672557985000 | {"id":14,"notes":["first note","second note"],"priority":1} | 000020420000244200002842 |
|
|
+---------------------+----+-------------------+-------------------------------------------------------------------------------+--------------------------+"#;
|
|
similar_asserts::assert_eq!(recordbatches.pretty_print().unwrap(), expected);
|
|
}
|
|
|
|
async fn verify_data_distribution(
|
|
instance: &MockDistributedInstance,
|
|
table_name: &str,
|
|
expected_distribution: HashMap<u32, &str>,
|
|
) {
|
|
let table = instance
|
|
.frontend()
|
|
.catalog_manager()
|
|
.table("greptime", "public", table_name, None)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let table_id = table.table_info().ident.table_id;
|
|
let table_route_value = instance
|
|
.table_metadata_manager()
|
|
.table_route_manager()
|
|
.table_route_storage()
|
|
.get(table_id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
|
|
let region_to_dn_map = region_distribution(
|
|
table_route_value
|
|
.region_routes()
|
|
.expect("physical table route"),
|
|
)
|
|
.iter()
|
|
.map(|(k, v)| (v.leader_regions[0], *k))
|
|
.collect::<HashMap<u32, u64>>();
|
|
assert!(region_to_dn_map.len() <= instance.datanodes().len());
|
|
|
|
let stmt = QueryLanguageParser::parse_sql(
|
|
&format!("SELECT ts, a, b FROM {table_name} ORDER BY ts"),
|
|
&QueryContext::arc(),
|
|
)
|
|
.unwrap();
|
|
let plan = instance
|
|
.frontend()
|
|
.statement_executor()
|
|
.plan(&stmt, QueryContext::arc())
|
|
.await
|
|
.unwrap();
|
|
let plan = DFLogicalSubstraitConvertor
|
|
.encode(&plan, DefaultSerializer)
|
|
.unwrap();
|
|
|
|
for (region, dn) in region_to_dn_map.iter() {
|
|
let region_server = instance.datanodes().get(dn).unwrap().region_server();
|
|
|
|
let region_id = RegionId::new(table_id, *region);
|
|
|
|
let stream = region_server
|
|
.handle_remote_read(
|
|
RegionQueryRequest {
|
|
region_id: region_id.as_u64(),
|
|
plan: plan.to_vec(),
|
|
..Default::default()
|
|
},
|
|
QueryContext::arc(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let actual = recordbatches.pretty_print().unwrap();
|
|
|
|
let expected = expected_distribution.get(region).unwrap();
|
|
assert_eq!(&actual, expected);
|
|
}
|
|
}
|
|
|
|
async fn test_insert_delete_and_query_on_auto_created_table(instance: &Instance) {
|
|
let insert = InsertRequest {
|
|
table_name: "auto_created_table".to_string(),
|
|
columns: vec![
|
|
Column {
|
|
column_name: "a".to_string(),
|
|
values: Some(Values {
|
|
i32_values: vec![4, 6],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![2],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Int32 as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "c".to_string(),
|
|
values: Some(Values {
|
|
string_values: vec![
|
|
r#"{ "id": 1, "name": "Alice", "age": 30, "active": true }"#
|
|
.to_string(),
|
|
r#"{ "id": 2, "name": "Bob", "balance": 1234.56, "active": false }"#
|
|
.to_string(),
|
|
],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![2],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Json as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: vec![
|
|
1672557975000,
|
|
1672557976000,
|
|
1672557977000,
|
|
],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count: 3,
|
|
};
|
|
|
|
// Test auto create not existed table upon insertion.
|
|
let request = Request::Inserts(InsertRequests {
|
|
inserts: vec![insert],
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(3)));
|
|
|
|
let insert = InsertRequest {
|
|
table_name: "auto_created_table".to_string(),
|
|
columns: vec![
|
|
Column {
|
|
column_name: "b".to_string(),
|
|
values: Some(Values {
|
|
string_values: vec!["x".to_string(), "z".to_string()],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![2],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::String as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: vec![
|
|
1672557978000,
|
|
1672557979000,
|
|
1672557980000,
|
|
],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count: 3,
|
|
};
|
|
|
|
// Test auto add not existed column upon insertion.
|
|
let request = Request::Inserts(InsertRequests {
|
|
inserts: vec![insert],
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(3)));
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::Sql(
|
|
"SELECT ts, a, b, json_to_string(c) as c FROM auto_created_table order by ts"
|
|
.to_string(),
|
|
)),
|
|
});
|
|
let output = query(instance, request.clone()).await;
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let expected = r#"+---------------------+---+---+--------------------------------------------------------+
|
|
| ts | a | b | c |
|
|
+---------------------+---+---+--------------------------------------------------------+
|
|
| 2023-01-01T07:26:15 | 4 | | {"active":true,"age":30,"id":1,"name":"Alice"} |
|
|
| 2023-01-01T07:26:16 | | | |
|
|
| 2023-01-01T07:26:17 | 6 | | {"active":false,"balance":1234.56,"id":2,"name":"Bob"} |
|
|
| 2023-01-01T07:26:18 | | x | |
|
|
| 2023-01-01T07:26:19 | | | |
|
|
| 2023-01-01T07:26:20 | | z | |
|
|
+---------------------+---+---+--------------------------------------------------------+"#;
|
|
similar_asserts::assert_eq!(recordbatches.pretty_print().unwrap(), expected);
|
|
|
|
let delete = DeleteRequest {
|
|
table_name: "auto_created_table".to_string(),
|
|
key_columns: vec![Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: vec![1672557975000, 1672557979000],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
}],
|
|
row_count: 2,
|
|
};
|
|
|
|
let output = query(
|
|
instance,
|
|
Request::Deletes(DeleteRequests {
|
|
deletes: vec![delete],
|
|
}),
|
|
)
|
|
.await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(2)));
|
|
|
|
let output = query(instance, request).await;
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let expected = r#"+---------------------+---+---+--------------------------------------------------------+
|
|
| ts | a | b | c |
|
|
+---------------------+---+---+--------------------------------------------------------+
|
|
| 2023-01-01T07:26:16 | | | |
|
|
| 2023-01-01T07:26:17 | 6 | | {"active":false,"balance":1234.56,"id":2,"name":"Bob"} |
|
|
| 2023-01-01T07:26:18 | | x | |
|
|
| 2023-01-01T07:26:20 | | z | |
|
|
+---------------------+---+---+--------------------------------------------------------+"#;
|
|
similar_asserts::assert_eq!(recordbatches.pretty_print().unwrap(), expected);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_promql_query() {
|
|
let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_promql_query")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
let table_name = "my_table";
|
|
let sql = format!(
|
|
"CREATE TABLE {table_name} (h string, a double, ts TIMESTAMP, TIME INDEX (ts), PRIMARY KEY(h))"
|
|
);
|
|
create_table(instance, sql).await;
|
|
|
|
let insert = InsertRequest {
|
|
table_name: table_name.to_string(),
|
|
columns: vec![
|
|
Column {
|
|
column_name: "h".to_string(),
|
|
values: Some(Values {
|
|
string_values: vec![
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
"t".to_string(),
|
|
],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Tag as i32,
|
|
datatype: ColumnDataType::String as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "a".to_string(),
|
|
values: Some(Values {
|
|
f64_values: vec![1f64, 11f64, 20f64, 22f64, 50f64, 55f64, 99f64],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![4],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Float64 as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: vec![
|
|
1672557972000,
|
|
1672557973000,
|
|
1672557974000,
|
|
1672557975000,
|
|
1672557976000,
|
|
1672557977000,
|
|
1672557978000,
|
|
1672557979000,
|
|
],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count: 8,
|
|
};
|
|
|
|
let request = Request::Inserts(InsertRequests {
|
|
inserts: vec![insert],
|
|
});
|
|
let output = query(instance, request).await;
|
|
assert!(matches!(output.data, OutputData::AffectedRows(8)));
|
|
|
|
let request = Request::Query(QueryRequest {
|
|
query: Some(Query::PromRangeQuery(api::v1::PromRangeQuery {
|
|
query: "my_table".to_owned(),
|
|
start: "1672557973".to_owned(),
|
|
end: "1672557978".to_owned(),
|
|
step: "1s".to_owned(),
|
|
lookback: "5m".to_string(),
|
|
})),
|
|
});
|
|
let output = query(instance, request).await;
|
|
let OutputData::Stream(stream) = output.data else {
|
|
unreachable!()
|
|
};
|
|
let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
|
|
let expected = "\
|
|
+---+------+---------------------+
|
|
| h | a | ts |
|
|
+---+------+---------------------+
|
|
| t | 11.0 | 2023-01-01T07:26:13 |
|
|
| t | | 2023-01-01T07:26:14 |
|
|
| t | 20.0 | 2023-01-01T07:26:15 |
|
|
| t | 22.0 | 2023-01-01T07:26:16 |
|
|
| t | 50.0 | 2023-01-01T07:26:17 |
|
|
| t | 55.0 | 2023-01-01T07:26:18 |
|
|
+---+------+---------------------+";
|
|
assert_eq!(recordbatches.pretty_print().unwrap(), expected);
|
|
}
|
|
|
|
#[apply(both_instances_cases)]
|
|
async fn test_extra_external_table_options(instance: Arc<dyn MockInstance>) {
|
|
common_telemetry::init_default_ut_logging();
|
|
let frontend = instance.frontend();
|
|
let instance = frontend.as_ref();
|
|
|
|
let insert = InsertRequest {
|
|
table_name: "auto_created_table".to_string(),
|
|
columns: vec![
|
|
Column {
|
|
column_name: "a".to_string(),
|
|
values: Some(Values {
|
|
i32_values: vec![4, 6],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![2],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Int32 as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "c".to_string(),
|
|
values: Some(Values {
|
|
string_values: vec![
|
|
r#"{ "id": 1, "name": "Alice", "age": 30, "active": true }"#
|
|
.to_string(),
|
|
r#"{ "id": 2, "name": "Bob", "balance": 1234.56, "active": false }"#
|
|
.to_string(),
|
|
],
|
|
..Default::default()
|
|
}),
|
|
null_mask: vec![2],
|
|
semantic_type: SemanticType::Field as i32,
|
|
datatype: ColumnDataType::Json as i32,
|
|
..Default::default()
|
|
},
|
|
Column {
|
|
column_name: "ts".to_string(),
|
|
values: Some(Values {
|
|
timestamp_millisecond_values: vec![
|
|
1672557975000,
|
|
1672557976000,
|
|
1672557977000,
|
|
],
|
|
..Default::default()
|
|
}),
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
..Default::default()
|
|
},
|
|
],
|
|
row_count: 3,
|
|
};
|
|
let request = Request::Inserts(InsertRequests {
|
|
inserts: vec![insert],
|
|
});
|
|
|
|
let ctx = Arc::new(
|
|
QueryContextBuilder::default()
|
|
.set_extension(TWCS_TIME_WINDOW.to_string(), "1d".to_string())
|
|
.build(),
|
|
);
|
|
let output = GrpcQueryHandler::do_query(instance, request, ctx)
|
|
.await
|
|
.unwrap();
|
|
assert!(matches!(output.data, OutputData::AffectedRows(3)));
|
|
|
|
let sql = "show create table auto_created_table";
|
|
let expected = r#"+--------------------+---------------------------------------------------+
|
|
| Table | Create Table |
|
|
+--------------------+---------------------------------------------------+
|
|
| auto_created_table | CREATE TABLE IF NOT EXISTS "auto_created_table" ( |
|
|
| | "a" INT NULL, |
|
|
| | "c" JSON NULL, |
|
|
| | "ts" TIMESTAMP(3) NOT NULL, |
|
|
| | TIME INDEX ("ts") |
|
|
| | ) |
|
|
| | |
|
|
| | ENGINE=mito |
|
|
| | WITH( |
|
|
| | 'comment' = 'Created on insertion', |
|
|
| | 'compaction.twcs.time_window' = '1d', |
|
|
| | 'compaction.type' = 'twcs' |
|
|
| | ) |
|
|
+--------------------+---------------------------------------------------+"#;
|
|
execute_sql_and_expect(&frontend, sql, expected).await;
|
|
}
|
|
|
|
fn declared_relationships_row_insert() -> Request {
|
|
use api::v1::value::ValueData;
|
|
use api::v1::{ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, Value};
|
|
|
|
let string_column = |name: &str, semantic: SemanticType| ColumnSchema {
|
|
column_name: name.to_string(),
|
|
datatype: ColumnDataType::String as i32,
|
|
semantic_type: semantic as i32,
|
|
..Default::default()
|
|
};
|
|
let string_value = |value: &str| Value {
|
|
value_data: Some(ValueData::StringValue(value.to_string())),
|
|
};
|
|
let mut schema = vec![ColumnSchema {
|
|
column_name: "observed_at".to_string(),
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
..Default::default()
|
|
}];
|
|
let mut values = vec![Value {
|
|
value_data: Some(ValueData::TimestampMillisecondValue(1000)),
|
|
}];
|
|
for (tag, value) in [
|
|
("src_type", "service"),
|
|
("src_id", "frontend"),
|
|
("rel_type", "depends_on"),
|
|
("dst_type", "service"),
|
|
("dst_id", "users-db"),
|
|
("provenance", "declared"),
|
|
("scope", ""),
|
|
("generation_id", ""),
|
|
] {
|
|
schema.push(string_column(tag, SemanticType::Tag));
|
|
values.push(string_value(value));
|
|
}
|
|
|
|
Request::RowInserts(RowInsertRequests {
|
|
inserts: vec![RowInsertRequest {
|
|
table_name: SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME.to_string(),
|
|
rows: Some(Rows {
|
|
schema,
|
|
rows: vec![Row { values }],
|
|
}),
|
|
}],
|
|
})
|
|
}
|
|
|
|
async fn assert_declared_relationships_table_is_canonical(instance: &Instance) {
|
|
let table = instance
|
|
.catalog_manager()
|
|
.table(
|
|
DEFAULT_CATALOG_NAME,
|
|
DEFAULT_PRIVATE_SCHEMA_NAME,
|
|
SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME,
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap()
|
|
.expect("declared-edge table must exist");
|
|
let table_info = table.table_info();
|
|
assert_eq!(table_info.meta.schema.column_schemas().len(), 20);
|
|
assert_eq!(
|
|
table_info
|
|
.meta
|
|
.schema
|
|
.timestamp_column()
|
|
.map(|c| c.name.as_str()),
|
|
Some("observed_at")
|
|
);
|
|
let primary_keys: Vec<&str> = table_info
|
|
.meta
|
|
.primary_key_indices
|
|
.iter()
|
|
.map(|idx| table_info.meta.schema.column_schemas()[*idx].name.as_str())
|
|
.collect();
|
|
assert_eq!(
|
|
primary_keys,
|
|
[
|
|
"src_type",
|
|
"src_id",
|
|
"rel_type",
|
|
"dst_type",
|
|
"dst_id",
|
|
"provenance",
|
|
"scope",
|
|
"generation_id",
|
|
]
|
|
);
|
|
}
|
|
|
|
/// The declared-edge table's first gRPC write must create it with the
|
|
/// canonical definition (never the request's shape), entering below the
|
|
/// user-DDL guard that rejects generic CREATE against the reserved name.
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_grpc_first_write_creates_declared_relationships_table() {
|
|
common_telemetry::init_default_ut_logging();
|
|
let standalone = GreptimeDbStandaloneBuilder::new("test_grpc_declared_first_write")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
let ctx = Arc::new(QueryContext::with(
|
|
DEFAULT_CATALOG_NAME,
|
|
DEFAULT_PRIVATE_SCHEMA_NAME,
|
|
));
|
|
let output =
|
|
GrpcQueryHandler::do_query(instance.as_ref(), declared_relationships_row_insert(), ctx)
|
|
.await
|
|
.unwrap();
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
|
|
assert_declared_relationships_table_is_canonical(instance).await;
|
|
}
|
|
|
|
/// The system-defined table is created regardless of the auto-create hint:
|
|
/// its creation is a system action, not user auto-create.
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn test_grpc_declared_relationships_bypasses_auto_create_hint() {
|
|
use api::v1::value::ValueData;
|
|
use api::v1::{ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, Value};
|
|
|
|
common_telemetry::init_default_ut_logging();
|
|
let standalone = GreptimeDbStandaloneBuilder::new("test_grpc_declared_no_auto_create")
|
|
.build()
|
|
.await;
|
|
let instance = standalone.fe_instance();
|
|
|
|
let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME);
|
|
ctx.set_extension(table::requests::AUTO_CREATE_TABLE_KEY, "false");
|
|
let ctx = Arc::new(ctx);
|
|
|
|
let output = GrpcQueryHandler::do_query(
|
|
instance.as_ref(),
|
|
declared_relationships_row_insert(),
|
|
ctx.clone(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert!(matches!(output.data, OutputData::AffectedRows(1)));
|
|
assert_declared_relationships_table_is_canonical(instance).await;
|
|
|
|
// An ordinary table stays subject to the hint.
|
|
let request = Request::RowInserts(RowInsertRequests {
|
|
inserts: vec![RowInsertRequest {
|
|
table_name: "ordinary_table".to_string(),
|
|
rows: Some(Rows {
|
|
schema: vec![ColumnSchema {
|
|
column_name: "ts".to_string(),
|
|
datatype: ColumnDataType::TimestampMillisecond as i32,
|
|
semantic_type: SemanticType::Timestamp as i32,
|
|
..Default::default()
|
|
}],
|
|
rows: vec![Row {
|
|
values: vec![Value {
|
|
value_data: Some(ValueData::TimestampMillisecondValue(1000)),
|
|
}],
|
|
}],
|
|
}),
|
|
}],
|
|
});
|
|
let err = GrpcQueryHandler::do_query(instance.as_ref(), request, ctx)
|
|
.await
|
|
.unwrap_err();
|
|
let msg = common_error::ext::ErrorExt::output_msg(&err);
|
|
assert!(msg.contains("auto_create_table"), "unexpected error: {msg}");
|
|
}
|
|
}
|