use crate::executor::base::ExecutionContext;
use crate::executor::build_error::PlanBuildError;
use crate::executor::streaming::operators::spec::{
DdlSpec, EdgeManageCommand, FulltextManageCommand, FulltextSpec, IndexManageCommand,
PropertyRename, SinkSpec, SpaceManageCommand, TagManageCommand, UserManageCommand,
VectorManageCommand, VectorSpec,
};
use linkrs_core::types::expr::Expression;
use super::contextual_to_expression;
pub(in crate::executor::streaming::plan::arena_builder) fn build_insert_vertices_spec(
node: &crate::planning::plan::core::nodes::data_modification::insert_nodes::InsertVerticesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let tag_property_names: Vec<String> = node.prop_names().to_vec();
Ok(SinkSpec::InsertVertices {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
vertex_properties: std::iter::once((
"vid".to_string(),
Expression::Variable("vid".to_string()),
))
.chain(
tag_property_names
.iter()
.map(|name| (name.clone(), Expression::Variable(name.clone()))),
)
.collect(),
tag: node.tag_name().to_string(),
tag_property_names,
if_not_exists: node.info().if_not_exists,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_insert_edges_spec(
node: &crate::planning::plan::core::nodes::data_modification::insert_nodes::InsertEdgesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
Ok(SinkSpec::InsertEdges {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
src_col: "src".to_string(),
dst_col: "dst".to_string(),
rank_col: "rank".to_string(),
edge_type: node.edge_name().to_string(),
edge_properties: node
.prop_names()
.iter()
.map(|name| (name.clone(), Expression::Variable(name.clone())))
.collect(),
if_not_exists: node.info().if_not_exists,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_delete_vertices_spec(
node: &crate::planning::plan::core::nodes::data_modification::delete_nodes::DeleteVerticesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let tag = node.tag().ok_or_else(|| {
PlanBuildError::missing_value(
"DeleteVertices",
node.id(),
"tag",
"DELETE VERTEX requires a tag qualifier",
)
})?;
Ok(SinkSpec::DeleteVertices {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
tag: tag.to_string(),
vertex_id_col: "vid".to_string(),
cascade: node.cascade(),
})
}
fn pipe_reference_column(expr: &linkrs_core::types::expr::ContextualExpression) -> Option<String> {
use linkrs_core::types::expr::Expression;
let inner = expr.expression()?;
match inner.inner() {
Expression::Variable(name) if name != "$-" => Some(name.clone()),
Expression::Property { object, property } => {
if let Expression::Variable(base) = object.as_ref() {
if base == "$-" {
return Some(property.clone());
}
}
Some(property.clone())
}
_ => None,
}
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_delete_edges_spec(
node: &crate::planning::plan::core::nodes::data_modification::delete_nodes::DeleteEdgesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let edge_type = node.edge_type().unwrap_or("").to_string();
Ok(SinkSpec::DeleteEdges {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
src_col: "src".to_string(),
dst_col: "dst".to_string(),
rank_col: "rank".to_string(),
edge_type,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_pipe_delete_vertices_spec(
node: &crate::planning::plan::core::nodes::data_modification::delete_nodes::PipeDeleteVerticesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let first = node.vertex_ids().first();
let vertex_id_col = first
.and_then(pipe_reference_column)
.unwrap_or_else(|| "vid".to_string());
let vertex_id_is_literal = first.and_then(pipe_reference_column).is_none();
Ok(SinkSpec::PipeDeleteVertices {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
vertex_id_col,
cascade: node.cascade(),
vertex_id_is_literal,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_pipe_delete_edges_spec(
node: &crate::planning::plan::core::nodes::data_modification::delete_nodes::PipeDeleteEdgesNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let edge_type = node.edge_type().unwrap_or("").to_string();
let (src_col, dst_col) = node
.edges()
.first()
.map(|(src, dst, _)| {
(
pipe_reference_column(src).unwrap_or_else(|| "src".to_string()),
pipe_reference_column(dst).unwrap_or_else(|| "dst".to_string()),
)
})
.unwrap_or_else(|| ("src".into(), "dst".to_string()));
Ok(SinkSpec::PipeDeleteEdges {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
src_col,
dst_col,
rank_col: "rank".to_string(),
edge_type,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_update_spec(
node: &crate::planning::plan::core::nodes::data_modification::update_nodes::UpdateNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
use crate::planning::plan::core::nodes::data_modification::info::UpdateTargetType;
match node.info() {
UpdateTargetType::Vertex(info) => Ok(SinkSpec::UpdateVertices {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
tag_name: info.tag_name.clone().unwrap_or_default(),
updates: info
.properties
.iter()
.filter_map(|(name, value)| value.get_expression().map(|expr| (name.clone(), expr)))
.collect(),
condition: info
.condition
.as_ref()
.map(contextual_to_expression)
.transpose()?,
is_upsert: info.is_upsert,
replace_properties: info.replace_properties,
}),
UpdateTargetType::Edge(info) => Ok(SinkSpec::UpdateEdges {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
src_col: "src".to_string(),
dst_col: "dst".to_string(),
rank_col: "rank".to_string(),
edge_type: info.edge_type.clone().unwrap_or_default(),
updates: info
.properties
.iter()
.filter_map(|(name, value)| value.get_expression().map(|expr| (name.clone(), expr)))
.collect(),
condition: info
.condition
.as_ref()
.map(contextual_to_expression)
.transpose()?,
is_upsert: info.is_upsert,
replace_properties: info.replace_properties,
}),
}
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_copy_from_spec(
node: &crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyFromNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let target = match node.target() {
crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyTarget::Vertex(
tag,
) => crate::executor::streaming::operators::spec::CopyTarget::Vertex(tag.clone()),
crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyTarget::Edge(
edge,
) => crate::executor::streaming::operators::spec::CopyTarget::Edge(edge.clone()),
};
Ok(SinkSpec::CopyFrom {
space_name: exec_ctx
.space_name
.clone()
.unwrap_or_else(|| node.space_name().to_string()),
target,
file_paths: node.file_paths().to_vec(),
by_column: node.by_column(),
header: node.header(),
delimiter: node.delimiter() as u8,
batch_size: node.batch_size(),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_copy_to_spec(
node: &crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyToNode,
exec_ctx: &ExecutionContext,
) -> Result<SinkSpec, PlanBuildError> {
let target = match node.target() {
crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyTarget::Vertex(
tag,
) => crate::executor::streaming::operators::spec::CopyTarget::Vertex(tag.clone()),
crate::planning::plan::core::nodes::data_modification::copy_nodes::CopyTarget::Edge(
edge,
) => crate::executor::streaming::operators::spec::CopyTarget::Edge(edge.clone()),
};
Ok(SinkSpec::CopyTo {
space_name: exec_ctx
.space_name
.clone()
.unwrap_or_else(|| node.space_name().to_string()),
target,
file_path: node.file_path().to_string(),
header: node.header(),
delimiter: node.delimiter() as u8,
})
}
fn space_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::SpaceManageNode,
) -> SpaceManageCommand {
use crate::executor::streaming::operators::spec::SpaceManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::SpaceManageNode::*;
match node {
Create(n) => SpaceManageCommand::Create {
space_name: n.info().space_name.clone(),
vid_type: n.info().vid_type.clone(),
comment: n.info().comment.clone(),
},
Drop(n) => SpaceManageCommand::Drop {
space_name: n.space_name().to_string(),
if_exists: n.if_exists(),
},
Desc(n) => SpaceManageCommand::Desc {
space_name: n.space_name().to_string(),
},
Show(_) => SpaceManageCommand::Show,
ShowCreate(n) => SpaceManageCommand::ShowCreate {
space_name: n.space_name().to_string(),
},
Switch(n) => SpaceManageCommand::Switch {
space_name: n.space_name().to_string(),
},
Clear(n) => SpaceManageCommand::Clear {
space_name: n.space_name().to_string(),
},
Checkpoint(_) => SpaceManageCommand::Checkpoint,
ExportDatabase(n) => SpaceManageCommand::ExportDatabase {
path: n.path().to_string(),
},
ImportDatabase(n) => SpaceManageCommand::ImportDatabase {
path: n.path().to_string(),
},
AttachDatabase(n) => SpaceManageCommand::AttachDatabase {
alias: n.alias().to_string(),
path: n.path().to_string(),
db_type: n.db_type().map(|s| s.to_string()),
},
DetachDatabase(n) => SpaceManageCommand::DetachDatabase {
alias: n.alias().to_string(),
},
PurgeExpired(n) => SpaceManageCommand::PurgeExpired {
space_name: n.space_name().to_string(),
target: n.target().cloned(),
},
}
}
fn tag_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::TagManageNode,
) -> TagManageCommand {
use crate::executor::streaming::operators::spec::TagManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::TagManageNode::*;
match node {
Create(n) => {
let info = n.info();
TagManageCommand::Create {
tag_name: info.tag_name.clone(),
properties: info.properties.clone(),
if_not_exists: info.if_not_exists,
ttl_duration: info.ttl_duration,
ttl_col: info.ttl_col.clone(),
}
}
Alter(n) => {
let info = n.info();
TagManageCommand::Alter {
tag_name: info.tag_name.clone(),
additions: info.additions.clone(),
deletions: info.deletions.clone(),
changes: info
.changes
.iter()
.map(|c| PropertyRename {
old_name: c.old_name.clone(),
new_name: c.new_name.clone(),
})
.collect(),
new_comment: info.new_comment.clone(),
}
}
Desc(n) => TagManageCommand::Desc {
tag_name: n.tag_name().to_string(),
},
Rename(n) => TagManageCommand::Rename {
old_name: n.old_name().to_string(),
new_name: n.new_name().to_string(),
},
Drop(n) => TagManageCommand::Drop {
tag_names: n.tag_names().to_vec(),
if_exists: n.if_exists(),
},
Show(_) => TagManageCommand::Show,
ShowCreate(n) => TagManageCommand::ShowCreate {
tag_name: n.tag_name().to_string(),
},
}
}
fn edge_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::EdgeManageNode,
) -> EdgeManageCommand {
use crate::executor::streaming::operators::spec::EdgeManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::EdgeManageNode::*;
match node {
Create(n) => {
let info = n.info();
EdgeManageCommand::Create {
edge_name: info.edge_name.clone(),
properties: info.properties.clone(),
src_tag_name: info.src_tag_name.clone(),
dst_tag_name: info.dst_tag_name.clone(),
if_not_exists: info.if_not_exists,
ttl_duration: info.ttl_duration,
ttl_col: info.ttl_col.clone(),
}
}
Alter(n) => {
let info = n.info();
EdgeManageCommand::Alter {
edge_name: info.edge_name.clone(),
additions: info.additions.clone(),
deletions: info.deletions.clone(),
changes: info
.changes
.iter()
.map(|c| PropertyRename {
old_name: c.old_name.clone(),
new_name: c.new_name.clone(),
})
.collect(),
new_comment: info.new_comment.clone(),
}
}
Rename(n) => EdgeManageCommand::Rename {
old_name: n.old_name().to_string(),
new_name: n.new_name().to_string(),
},
UpdateEndpoints(n) => EdgeManageCommand::UpdateEndpoints {
edge_name: n.edge_name().to_string(),
src_tag_name: n.src_tag_name().to_string(),
dst_tag_name: n.dst_tag_name().to_string(),
clear_constraint: n.clear_constraint(),
},
Desc(n) => EdgeManageCommand::Desc {
edge_name: n.edge_name().to_string(),
},
Drop(n) => EdgeManageCommand::Drop {
edge_names: n.edge_names().to_vec(),
if_exists: n.if_exists(),
},
Show(_) => EdgeManageCommand::Show,
ShowCreate(n) => EdgeManageCommand::ShowCreate {
edge_name: n.edge_name().to_string(),
},
}
}
fn index_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::IndexManageNode,
) -> IndexManageCommand {
use crate::executor::streaming::operators::spec::IndexManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::IndexManageNode::*;
match node {
CreateTagIndex(n) => {
let info = n.info();
IndexManageCommand::CreateTagIndex {
index_name: info.index_name.clone(),
target_name: info.target_name.clone(),
properties: info.properties.clone(),
}
}
DropTagIndex(n) => IndexManageCommand::DropTagIndex {
index_name: n.index_name().to_string(),
if_exists: n.if_exists(),
},
DescTagIndex(n) => IndexManageCommand::DescTagIndex {
index_name: n.index_name().to_string(),
},
ShowTagIndexes(_) => IndexManageCommand::ShowTagIndexes,
RebuildTagIndex(n) => IndexManageCommand::RebuildTagIndex {
index_name: n.index_name().to_string(),
},
CreateEdgeIndex(n) => {
let info = n.info();
IndexManageCommand::CreateEdgeIndex {
index_name: info.index_name.clone(),
target_name: info.target_name.clone(),
properties: info.properties.clone(),
}
}
DropEdgeIndex(n) => IndexManageCommand::DropEdgeIndex {
index_name: n.index_name().to_string(),
if_exists: n.if_exists(),
},
DescEdgeIndex(n) => IndexManageCommand::DescEdgeIndex {
index_name: n.index_name().to_string(),
},
ShowEdgeIndexes(_) => IndexManageCommand::ShowEdgeIndexes,
RebuildEdgeIndex(n) => IndexManageCommand::RebuildEdgeIndex {
index_name: n.index_name().to_string(),
},
ShowIndexes(_) => IndexManageCommand::ShowIndexes,
ShowCreateIndex(n) => IndexManageCommand::ShowCreateIndex {
index_name: n.index_name().to_string(),
},
}
}
fn user_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::UserManageNode,
) -> UserManageCommand {
use crate::executor::streaming::operators::spec::UserManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::UserManageNode::*;
match node {
Create(n) => UserManageCommand::Create {
username: n.username().to_string(),
password: n.password().to_string(),
if_not_exists: n.if_not_exists(),
},
Alter(n) => UserManageCommand::Alter {
username: n.username().to_string(),
new_password: n.new_password().cloned(),
is_locked: n.is_locked(),
history_limit: n.password_history_depth(),
},
Drop(n) => UserManageCommand::Drop {
username: n.username().to_string(),
if_exists: n.if_exists(),
},
ChangePassword(n) => UserManageCommand::ChangePassword {
password_info: n.password_info().clone(),
},
GrantRole(n) => UserManageCommand::GrantRole {
username: n.username().to_string(),
space_name: n.space_name().to_string(),
role: n.role().to_string(),
},
RevokeRole(n) => UserManageCommand::RevokeRole {
username: n.username().to_string(),
space_name: n.space_name().to_string(),
role: n.role().to_string(),
},
ShowUsers(_) => UserManageCommand::ShowUsers,
ShowRoles(n) => UserManageCommand::ShowRoles {
space_name: n.space_name().to_string(),
},
DescribeUser(n) => UserManageCommand::DescribeUser {
username: n.username().to_string(),
},
}
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_space_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::SpaceManageNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::SpaceManage {
command: space_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_tag_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::TagManageNode,
exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::TagManage {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
command: tag_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_edge_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::EdgeManageNode,
exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::EdgeManage {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
command: edge_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_index_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::IndexManageNode,
exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::IndexManage {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
command: index_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_delete_index_spec(
node: &crate::planning::plan::core::nodes::data_modification::delete_nodes::DeleteIndexNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::DeleteIndex {
space_name: node.info().space_name.clone(),
index_name: node.info().index_name.clone(),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_user_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::UserManageNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
Ok(DdlSpec::UserManage {
command: user_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_macro_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::MacroManageNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
use crate::executor::streaming::operators::spec::MacroManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::MacroManageNode::*;
let command = match node {
Create(n) => MacroManageCommand::Create {
macro_name: n.macro_name().to_string(),
params: n.info().params.clone(),
body: n.info().body.clone(),
if_not_exists: n.info().if_not_exists,
},
Drop(n) => MacroManageCommand::Drop {
macro_name: n.macro_name().to_string(),
if_exists: n.if_exists(),
},
};
Ok(DdlSpec::MacroManage { command })
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_type_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::TypeManageNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
use crate::executor::streaming::operators::spec::TypeManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::TypeManageNode::*;
let command = match node {
Create(n) => TypeManageCommand::Create {
type_name: n.alias_name().to_string(),
underlying_type: n.info().underlying_type.clone(),
if_not_exists: n.info().if_not_exists,
},
Drop(n) => TypeManageCommand::Drop {
type_name: n.alias_name().to_string(),
if_exists: n.if_exists(),
},
};
Ok(DdlSpec::TypeManage { command })
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_sequence_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::SequenceManageNode,
_exec_ctx: &ExecutionContext,
) -> Result<DdlSpec, PlanBuildError> {
use crate::executor::streaming::operators::spec::SequenceManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::SequenceManageNode::*;
let command = match node {
Create(n) => SequenceManageCommand::Create {
seq_name: n.seq_name().to_string(),
start: n.info().start,
increment: n.info().increment,
min_value: n.info().min_value,
max_value: n.info().max_value,
cycle: n.info().cycle.unwrap_or(false),
if_not_exists: n.info().if_not_exists,
},
Alter(n) => SequenceManageCommand::Alter {
seq_name: n.seq_name().to_string(),
increment: n.info().increment,
min_value: n.info().min_value,
max_value: n.info().max_value,
cycle: n.info().cycle,
},
Drop(n) => SequenceManageCommand::Drop {
seq_name: n.seq_name().to_string(),
if_exists: n.if_exists(),
},
};
Ok(DdlSpec::SequenceManage { command })
}
fn fulltext_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::FulltextManageNode,
) -> FulltextManageCommand {
use crate::executor::streaming::operators::spec::FulltextManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::FulltextManageNode::*;
match node {
Create(n) => FulltextManageCommand::Create {
index_name: n.index_name.clone(),
schema_name: n.schema_name.clone(),
fields: n.fields.clone(),
engine_type: n.engine_type,
options: n.options.clone(),
if_not_exists: n.if_not_exists,
space_id: n.space_id,
},
Drop(n) => FulltextManageCommand::Drop {
index_name: n.index_name.clone(),
if_exists: n.if_exists,
},
Alter(n) => FulltextManageCommand::Alter {
index_name: n.index_name.clone(),
actions: n.actions.clone(),
},
Show(n) => FulltextManageCommand::Show {
pattern: n.pattern.clone(),
from_schema: n.from_schema.clone(),
},
Describe(n) => FulltextManageCommand::Describe {
index_name: n.index_name.clone(),
},
}
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_fulltext_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::FulltextManageNode,
exec_ctx: &ExecutionContext,
) -> Result<FulltextSpec, PlanBuildError> {
Ok(FulltextSpec::FulltextManage {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
command: fulltext_manage_to_command(node),
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_fulltext_search_spec(
node: &crate::planning::plan::core::nodes::search::fulltext::data_access::FulltextSearchNode,
exec_ctx: &ExecutionContext,
) -> Result<FulltextSpec, PlanBuildError> {
Ok(FulltextSpec::FulltextSearch {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
space_id: exec_ctx.current_space_id().unwrap_or(0),
index_name: node.index_name.clone(),
structured_query: fulltext_query_to_structured(&node.query, &node.field_name)?,
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
limit: node.limit,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_fulltext_lookup_spec(
node: &crate::planning::plan::core::nodes::search::fulltext::data_access::FulltextLookupNode,
exec_ctx: &ExecutionContext,
) -> Result<FulltextSpec, PlanBuildError> {
Ok(FulltextSpec::FulltextLookup {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
space_id: exec_ctx.current_space_id().unwrap_or(0),
index_name: node.index_name.clone(),
structured_query: linkrs_fulltext::query::FulltextQuery::Simple(node.query.clone()),
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
limit: node.limit,
})
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_match_fulltext_spec(
node: &crate::planning::plan::core::nodes::search::fulltext::data_access::MatchFulltextNode,
exec_ctx: &ExecutionContext,
) -> Result<FulltextSpec, PlanBuildError> {
Ok(FulltextSpec::MatchFulltext {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
space_id: node.space_id,
match_expr: Expression::Literal(linkrs_core::Value::string(format!(
"{}:{}",
node.fulltext_condition.field, node.fulltext_condition.query
))),
match_field: Some(node.field_name.clone()),
structured_query: linkrs_fulltext::query::FulltextQuery::Simple(
node.fulltext_condition.query.clone(),
),
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
limit: None,
})
}
fn vector_manage_to_command(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::VectorManageNode,
) -> VectorManageCommand {
use crate::executor::streaming::operators::spec::VectorManageCommand;
use crate::planning::plan::core::nodes::management::manage_node_enums::VectorManageNode::*;
match node {
Create(n) => VectorManageCommand::Create {
index_name: n.index_name.clone(),
tag_name: n.tag_name.clone(),
field_name: n.field_name.clone(),
vector_size: n.vector_size,
distance: n.distance,
space_id: n.space_id,
hnsw_m: n.hnsw_m,
hnsw_ef_construct: n.hnsw_ef_construct,
quantization: n.quantization,
quantile: n.quantile,
compression: n.compression,
always_ram: n.always_ram,
},
Drop(n) => VectorManageCommand::Drop {
index_name: n.index_name.clone(),
if_exists: n.if_exists,
space_id: n.space_id,
tag_name: n.tag_name.clone(),
field_name: n.field_name.clone(),
},
}
}
pub(in crate::executor::streaming::plan::arena_builder) fn build_vector_manage_spec(
node: &crate::planning::plan::core::nodes::management::manage_node_enums::VectorManageNode,
exec_ctx: &ExecutionContext,
) -> Result<VectorSpec, PlanBuildError> {
Ok(VectorSpec::VectorManage {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
command: vector_manage_to_command(node),
})
}
#[cfg(feature = "vector")]
pub(in crate::executor::streaming::plan::arena_builder) fn build_simvec_spec(
node: &crate::planning::plan::core::nodes::search::vector::data_access::VectorSearchNode,
exec_ctx: &ExecutionContext,
) -> Result<VectorSpec, PlanBuildError> {
let (query_vector, query_text) = vector_query_to_vec(&node.query, exec_ctx)?;
Ok(VectorSpec::VectorSearch {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
space_id: node.space_id,
index_name: node.index_name.clone(),
query_vector,
query_text,
top_k: node.limit as u32,
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
threshold: node.threshold,
filter: node.filter.clone(),
offset: node.offset,
ryw_config: exec_ctx.ryw_config,
})
}
#[cfg(feature = "vector")]
pub(in crate::executor::streaming::plan::arena_builder) fn build_vector_lookup_spec(
node: &crate::planning::plan::core::nodes::search::vector::data_access::VectorLookupNode,
exec_ctx: &ExecutionContext,
) -> Result<VectorSpec, PlanBuildError> {
let (query_vector, query_text) = vector_query_to_vec(&node.query, exec_ctx)?;
Ok(VectorSpec::VectorLookup {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
space_id: node.space_id,
index_name: node.index_name.clone(),
query_vector,
query_text,
top_k: node.limit as u32,
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
ryw_config: exec_ctx.ryw_config,
})
}
#[cfg(feature = "vector")]
pub(in crate::executor::streaming::plan::arena_builder) fn build_vector_match_spec(
node: &crate::planning::plan::core::nodes::search::vector::data_access::VectorMatchNode,
exec_ctx: &ExecutionContext,
) -> Result<VectorSpec, PlanBuildError> {
let (query_vector, query_text) = vector_query_to_vec(&node.query, exec_ctx)?;
Ok(VectorSpec::VectorMatch {
space_name: exec_ctx.space_name.clone().unwrap_or_default(),
pattern: node.pattern.clone(),
field: node.field.clone(),
query_vector,
query_text,
threshold: node.threshold,
tag_name: node.tag_name.clone(),
field_name: node.field_name.clone(),
space_id: node.space_id,
ryw_config: exec_ctx.ryw_config,
})
}
#[cfg(feature = "vector")]
fn vector_query_to_vec(
expr: &crate::parser::ast::vector::VectorQueryExpr,
exec_ctx: &ExecutionContext,
) -> Result<(Vec<f32>, Option<String>), PlanBuildError> {
match expr.query_type {
crate::parser::ast::vector::VectorQueryType::Vector => {
let vec: Vec<f32> = serde_json::from_str(&expr.query_data).unwrap_or_default();
Ok((vec, None))
}
crate::parser::ast::vector::VectorQueryType::Text => {
Ok((vec![], Some(expr.query_data.clone())))
}
crate::parser::ast::vector::VectorQueryType::Parameter => {
let param_name = expr.query_data.trim_start_matches('$');
let val = exec_ctx.get_param(param_name).ok_or_else(|| {
PlanBuildError::capability(
"parameterized-vector-query",
format!("Parameter '${}' is not bound", param_name),
)
})?;
let vec = resolve_param_to_vector(val, param_name)?;
Ok((vec, None))
}
}
}
#[cfg(feature = "vector")]
fn resolve_param_to_vector(
val: &linkrs_core::Value,
param_name: &str,
) -> Result<Vec<f32>, PlanBuildError> {
match val {
linkrs_core::Value::String(s) => serde_json::from_str(s).map_err(|_| {
PlanBuildError::capability(
"parameterized-vector-query",
format!(
"Parameter '${}' value is not a valid JSON vector: '{}'",
param_name, s
),
)
}),
linkrs_core::Value::List(list) => list
.iter()
.map(|v| match v {
linkrs_core::Value::Float(f) => Ok(*f),
linkrs_core::Value::Int(i) => Ok(*i as f32),
linkrs_core::Value::Double(d) => Ok(*d as f32),
_ => Err(PlanBuildError::capability(
"parameterized-vector-query",
format!(
"Parameter '${}' element is not numeric: {:?}",
param_name, v
),
)),
})
.collect(),
_ => Err(PlanBuildError::capability(
"parameterized-vector-query",
format!(
"Parameter '${}' value is not a string or list: {:?}",
param_name, val
),
)),
}
}
fn fulltext_query_to_structured(
expr: &crate::parser::ast::fulltext::FulltextQueryExpr,
expected_field: &str,
) -> Result<linkrs_fulltext::query::FulltextQuery, PlanBuildError> {
use crate::parser::ast::fulltext::FulltextQueryExpr;
use linkrs_fulltext::query::FulltextQuery;
match expr {
FulltextQueryExpr::Simple(text) => Ok(FulltextQuery::Simple(text.clone())),
FulltextQueryExpr::Phrase(text) => Ok(FulltextQuery::Phrase(text.clone())),
FulltextQueryExpr::Prefix(text) => Ok(FulltextQuery::Prefix(text.clone())),
FulltextQueryExpr::Fuzzy(text, distance) => {
Ok(FulltextQuery::Fuzzy(text.clone(), *distance))
}
FulltextQueryExpr::Wildcard(text) => Ok(FulltextQuery::Wildcard(text.clone())),
FulltextQueryExpr::Boolean {
must,
should,
must_not,
} => {
let convert = |items: &[FulltextQueryExpr]| {
items
.iter()
.map(|q| fulltext_query_to_structured(q, expected_field))
.collect::<Result<Vec<_>, _>>()
};
Ok(FulltextQuery::Boolean {
must: convert(must)?,
should: convert(should)?,
must_not: convert(must_not)?,
})
}
FulltextQueryExpr::Range {
field,
lower,
upper,
include_lower,
include_upper,
} => {
if !field.is_empty() && field != expected_field {
return Err(PlanBuildError::capability(
"fulltext-field-mismatch",
format!(
"Range query targets field '{}' but index serves field '{}'",
field, expected_field
),
));
}
Ok(FulltextQuery::Range {
lower: lower.clone(),
upper: upper.clone(),
include_lower: *include_lower,
include_upper: *include_upper,
})
}
FulltextQueryExpr::Field(field, text) => {
if field != expected_field {
return Err(PlanBuildError::capability(
"fulltext-field-mismatch",
format!(
"Query targets field '{}' but index serves field '{}'",
field, expected_field
),
));
}
Ok(FulltextQuery::Simple(text.clone()))
}
FulltextQueryExpr::MultiField(fields) => {
let mut matched = Vec::with_capacity(fields.len());
for (field, text) in fields {
if field != expected_field {
return Err(PlanBuildError::capability(
"fulltext-field-mismatch",
format!(
"Multi-field query targets field '{}' but index serves field '{}'",
field, expected_field
),
));
}
matched.push(FulltextQuery::Simple(text.clone()));
}
Ok(FulltextQuery::Boolean {
must: vec![],
should: matched,
must_not: vec![],
})
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::executor::base::ExecutionContext;
use crate::parser::ast::fulltext::{BM25Options, IndexFieldDef, IndexOptions};
use crate::planning::plan::core::nodes::management::manage_node_enums::{
FulltextManageNode, SequenceManageNode,
};
use crate::planning::plan::core::nodes::management::sequence_nodes::{
AlterSequenceNode, CreateSequenceNode, DropSequenceNode, SequenceManageInfo,
};
use crate::planning::plan::core::nodes::search::fulltext::management::CreateFulltextIndexNode;
use linkrs_core::types::FulltextEngineType;
use std::collections::HashMap;
fn exec_ctx() -> ExecutionContext {
ExecutionContext::default()
}
#[test]
fn sequence_create_spec_carries_all_parameters() {
let info = SequenceManageInfo::new("s".to_string())
.with_start(Some(1))
.with_increment(Some(2))
.with_bounds(Some(1), Some(100))
.with_cycle(Some(true))
.with_if_not_exists(true);
let node = SequenceManageNode::Create(CreateSequenceNode::new(1, info));
match build_sequence_manage_spec(&node, &exec_ctx()).expect("spec builds") {
DdlSpec::SequenceManage {
command:
crate::executor::streaming::operators::spec::SequenceManageCommand::Create {
seq_name,
start,
increment,
min_value,
max_value,
cycle,
if_not_exists,
},
} => {
assert_eq!(seq_name, "s");
assert_eq!(start, Some(1));
assert_eq!(increment, Some(2));
assert_eq!(min_value, Some(1));
assert_eq!(max_value, Some(100));
assert!(cycle);
assert!(if_not_exists);
}
other => panic!("expected SequenceManage::Create, got {other:?}"),
}
}
#[test]
fn sequence_alter_and_drop_specs_carry_flags() {
let info = SequenceManageInfo::new("s".to_string())
.with_increment(Some(5))
.with_cycle(Some(false));
let node = SequenceManageNode::Alter(AlterSequenceNode::new(2, info));
match build_sequence_manage_spec(&node, &exec_ctx()).expect("spec builds") {
DdlSpec::SequenceManage {
command:
crate::executor::streaming::operators::spec::SequenceManageCommand::Alter {
seq_name,
increment,
cycle,
..
},
} => {
assert_eq!(seq_name, "s");
assert_eq!(increment, Some(5));
assert_eq!(cycle, Some(false));
}
other => panic!("expected SequenceManage::Alter, got {other:?}"),
}
let node = SequenceManageNode::Drop(DropSequenceNode::new(3, "s".to_string()));
match build_sequence_manage_spec(&node, &exec_ctx()).expect("spec builds") {
DdlSpec::SequenceManage {
command:
crate::executor::streaming::operators::spec::SequenceManageCommand::Drop {
seq_name,
if_exists,
},
} => {
assert_eq!(seq_name, "s");
assert!(!if_exists);
}
other => panic!("expected SequenceManage::Drop, got {other:?}"),
}
}
#[test]
fn fulltext_create_spec_keeps_engine_fields_options_and_flag() {
let fields = vec![IndexFieldDef {
field_name: "content".to_string(),
analyzer: None,
boost: None,
}];
let options = IndexOptions {
bm25_config: Some(BM25Options {
k1: Some(1.2),
b: Some(0.8),
field_weights: HashMap::new(),
analyzer: None,
store_original: None,
}),
common_options: HashMap::new(),
};
let plan_node = CreateFulltextIndexNode::new(
"idx_a".to_string(),
"Article".to_string(),
fields,
FulltextEngineType::Bm25,
options,
true,
7,
);
let node = FulltextManageNode::Create(plan_node);
match fulltext_manage_to_command(&node) {
FulltextManageCommand::Create {
index_name,
schema_name,
fields,
engine_type,
options,
if_not_exists,
space_id,
} => {
assert_eq!(index_name, "idx_a");
assert_eq!(schema_name, "Article");
assert_eq!(fields.len(), 1);
assert_eq!(fields[0].field_name, "content");
assert_eq!(engine_type, FulltextEngineType::Bm25);
let bm25 = options.bm25_config.expect("bm25 options carried");
assert_eq!(bm25.k1, Some(1.2));
assert_eq!(bm25.b, Some(0.8));
assert!(if_not_exists);
assert_eq!(space_id, 7);
}
other => panic!("expected FulltextManageCommand::Create, got {other:?}"),
}
}
}