mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-07-06 22:10:39 +00:00
feat: identify noneffective binary modifiers (#8230)
* perf(promql): use tsid for full-label modifier joins * fix(promql): collect narrow range joins by tsid * verify on for both sides Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * check if tsid exists Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * update sqlness Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * correct behavior Signed-off-by: Ruihang Xia <waynestxia@gmail.com> --------- Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
This commit is contained in:
Ruihang Xia
GitHub
@@ -82,20 +82,19 @@ impl PromqlTsidNarrowJoin {
|
||||
}
|
||||
|
||||
fn is_promql_value_tsid_time_schema(schema: &SchemaRef) -> bool {
|
||||
let mut has_value = false;
|
||||
let mut value_columns = 0;
|
||||
let mut has_tsid = false;
|
||||
let mut has_time = false;
|
||||
|
||||
for field in schema.fields() {
|
||||
match field.name().as_str() {
|
||||
"greptime_value" => has_value = true,
|
||||
DATA_SCHEMA_TSID_COLUMN_NAME => has_tsid = true,
|
||||
_ if matches!(field.data_type(), DataType::Timestamp(_, _)) => has_time = true,
|
||||
_ => return false,
|
||||
_ => value_columns += 1,
|
||||
}
|
||||
}
|
||||
|
||||
has_value && has_tsid && has_time
|
||||
value_columns == 1 && has_tsid && has_time
|
||||
}
|
||||
|
||||
fn joins_on_tsid_and_time(hash_join: &HashJoinExec) -> bool {
|
||||
@@ -214,6 +213,60 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chooses_collect_left_for_computed_narrow_value_column() {
|
||||
let left = Arc::new(EmptyExec::new(Arc::new(Schema::new(vec![
|
||||
Field::new("prom_rate(greptime_value)", DataType::Float64, true),
|
||||
Field::new(DATA_SCHEMA_TSID_COLUMN_NAME, DataType::UInt64, false),
|
||||
Field::new(
|
||||
"greptime_timestamp",
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
])))) as Arc<dyn ExecutionPlan>;
|
||||
let right = Arc::new(EmptyExec::new(Arc::new(Schema::new(vec![
|
||||
Field::new("greptime_value", DataType::Float64, true),
|
||||
Field::new("host", DataType::Utf8, true),
|
||||
Field::new(DATA_SCHEMA_TSID_COLUMN_NAME, DataType::UInt64, false),
|
||||
Field::new(
|
||||
"greptime_timestamp",
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
])))) as Arc<dyn ExecutionPlan>;
|
||||
let on = vec![
|
||||
(
|
||||
Arc::new(Column::new(DATA_SCHEMA_TSID_COLUMN_NAME, 1)) as Arc<dyn PhysicalExpr>,
|
||||
Arc::new(Column::new(DATA_SCHEMA_TSID_COLUMN_NAME, 2)) as Arc<dyn PhysicalExpr>,
|
||||
),
|
||||
(
|
||||
Arc::new(Column::new("greptime_timestamp", 2)) as Arc<dyn PhysicalExpr>,
|
||||
Arc::new(Column::new("greptime_timestamp", 3)) as Arc<dyn PhysicalExpr>,
|
||||
),
|
||||
];
|
||||
let join = Arc::new(
|
||||
HashJoinExec::try_new(
|
||||
left,
|
||||
right,
|
||||
on,
|
||||
None,
|
||||
&JoinType::Inner,
|
||||
Some(vec![0, 3, 4, 5, 6]),
|
||||
PartitionMode::Partitioned,
|
||||
NullEquality::NullEqualsNull,
|
||||
false,
|
||||
)
|
||||
.unwrap(),
|
||||
) as Arc<dyn ExecutionPlan>;
|
||||
|
||||
let optimized = PromqlTsidNarrowJoin
|
||||
.optimize(join, &ConfigOptions::default())
|
||||
.unwrap();
|
||||
let optimized_join = optimized.as_any().downcast_ref::<HashJoinExec>().unwrap();
|
||||
|
||||
assert_eq!(optimized_join.partition_mode(), &PartitionMode::CollectLeft);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keeps_partitioned_join_when_left_side_carries_labels() {
|
||||
let left = Arc::new(EmptyExec::new(Arc::new(Schema::new(vec![
|
||||
|
||||
@@ -876,14 +876,14 @@ impl PromPlanner {
|
||||
) -> Result<LogicalPlan> {
|
||||
let only_join_time_index =
|
||||
first_leaf.ctx.tag_columns.is_empty() || right_leaf.ctx.tag_columns.is_empty();
|
||||
let (mut left_keys, mut right_keys) = Self::binary_join_key_columns(
|
||||
&left,
|
||||
&right_leaf.plan,
|
||||
let (mut left_keys, mut right_keys, force_empty_join) = self.binary_join_key_columns(
|
||||
left.schema(),
|
||||
right_leaf.plan.schema(),
|
||||
&first_leaf.ctx,
|
||||
&right_leaf.ctx,
|
||||
only_join_time_index,
|
||||
&None,
|
||||
);
|
||||
)?;
|
||||
|
||||
if let (Some(left_time_index_column), Some(right_time_index_column)) = (
|
||||
first_leaf.ctx.time_index_column.clone(),
|
||||
@@ -907,7 +907,7 @@ impl PromPlanner {
|
||||
.map(|name| Column::new(Some(right_leaf.alias.clone()), name))
|
||||
.collect::<Vec<_>>(),
|
||||
),
|
||||
None,
|
||||
force_empty_join.then_some(lit(false)),
|
||||
NullEquality::NullEqualsNull,
|
||||
)
|
||||
.context(DataFusionPlanningSnafu)?
|
||||
@@ -1250,6 +1250,14 @@ impl PromPlanner {
|
||||
}
|
||||
let (output_field_columns, field_columns) =
|
||||
Self::align_binary_field_columns(&left_field_columns, &right_field_columns);
|
||||
let left_aligned_field_columns = field_columns
|
||||
.iter()
|
||||
.map(|(left_col_name, _)| (*left_col_name).clone())
|
||||
.collect::<Vec<_>>();
|
||||
let right_aligned_field_columns = field_columns
|
||||
.iter()
|
||||
.map(|(_, right_col_name)| (*right_col_name).clone())
|
||||
.collect::<Vec<_>>();
|
||||
// PromQL binary arithmetic only combines the shared prefix of value columns.
|
||||
// Keep the output field count aligned with that zipped prefix so planning
|
||||
// remains stable even when the two sides have uneven multi-field schemas.
|
||||
@@ -1267,6 +1275,8 @@ impl PromPlanner {
|
||||
// under this case we only join on time index
|
||||
left_context.tag_columns.is_empty() || right_context.tag_columns.is_empty(),
|
||||
modifier,
|
||||
&left_context,
|
||||
&right_context,
|
||||
)?;
|
||||
let join_plan_schema = join_plan.schema().clone();
|
||||
|
||||
@@ -1300,14 +1310,21 @@ impl PromPlanner {
|
||||
// So we filter on the join result and then project only the side that should
|
||||
// be preserved according to PromQL semantics.
|
||||
let filtered = self.filter_on_field_column(join_plan, bin_expr_builder)?;
|
||||
let (project_table_ref, project_context) =
|
||||
let (project_table_ref, mut project_context, project_field_columns) =
|
||||
match (lhs.value_type(), rhs.value_type()) {
|
||||
(ValueType::Scalar, ValueType::Vector) => {
|
||||
(&right_table_ref, &right_context)
|
||||
}
|
||||
_ => (&left_table_ref, &left_context),
|
||||
(ValueType::Scalar, ValueType::Vector) => (
|
||||
&right_table_ref,
|
||||
right_context.clone(),
|
||||
right_aligned_field_columns,
|
||||
),
|
||||
_ => (
|
||||
&left_table_ref,
|
||||
left_context.clone(),
|
||||
left_aligned_field_columns,
|
||||
),
|
||||
};
|
||||
self.project_binary_join_side(filtered, project_table_ref, project_context)
|
||||
project_context.field_columns = project_field_columns;
|
||||
self.project_binary_join_side(filtered, project_table_ref, &project_context)
|
||||
} else {
|
||||
self.projection_for_each_field_column(join_plan, bin_expr_builder)
|
||||
}
|
||||
@@ -1600,6 +1617,11 @@ impl PromPlanner {
|
||||
self.create_function_expr(func, args.literals.clone(), query_engine_state)?;
|
||||
func_exprs.insert(0, self.create_time_index_column_expr()?);
|
||||
func_exprs.extend_from_slice(&self.create_tag_column_exprs()?);
|
||||
if let Some(tsid_col) =
|
||||
Self::optional_tsid_projection(input.schema(), None, self.ctx.use_tsid)
|
||||
{
|
||||
func_exprs.push(tsid_col);
|
||||
}
|
||||
|
||||
let builder = LogicalPlanBuilder::from(input)
|
||||
.project(func_exprs)
|
||||
@@ -2848,6 +2870,7 @@ impl PromPlanner {
|
||||
}
|
||||
|
||||
"label_join" => {
|
||||
self.ctx.use_tsid = false;
|
||||
let (concat_expr, dst_label) = Self::build_concat_labels_expr(
|
||||
&mut other_input_exprs,
|
||||
&self.ctx,
|
||||
@@ -2871,6 +2894,7 @@ impl PromPlanner {
|
||||
ScalarFunc::GeneratedExpr
|
||||
}
|
||||
"label_replace" => {
|
||||
self.ctx.use_tsid = false;
|
||||
if let Some((replace_expr, dst_label)) = self
|
||||
.build_regexp_replace_label_expr(&mut other_input_exprs, query_engine_state)?
|
||||
{
|
||||
@@ -3978,20 +4002,26 @@ impl PromPlanner {
|
||||
}
|
||||
|
||||
fn binary_join_key_columns(
|
||||
left: &LogicalPlan,
|
||||
right: &LogicalPlan,
|
||||
left_ctx: &PromPlannerContext,
|
||||
right_ctx: &PromPlannerContext,
|
||||
&self,
|
||||
left_schema: &DFSchemaRef,
|
||||
right_schema: &DFSchemaRef,
|
||||
left_context: &PromPlannerContext,
|
||||
right_context: &PromPlannerContext,
|
||||
only_join_time_index: bool,
|
||||
modifier: &Option<BinModifier>,
|
||||
) -> (BTreeSet<String>, BTreeSet<String>) {
|
||||
) -> Result<(BTreeSet<String>, BTreeSet<String>, bool)> {
|
||||
let has_tsid = |schema: &DFSchemaRef| {
|
||||
schema
|
||||
.fields()
|
||||
.iter()
|
||||
.any(|field| field.name() == DATA_SCHEMA_TSID_COLUMN_NAME)
|
||||
};
|
||||
let use_tsid_join = !only_join_time_index
|
||||
&& modifier.as_ref().is_none_or(|modifier| {
|
||||
modifier.matching.is_none()
|
||||
&& matches!(modifier.card, VectorMatchCardinality::OneToOne)
|
||||
})
|
||||
&& Self::plan_has_tsid_column(left)
|
||||
&& Self::plan_has_tsid_column(right);
|
||||
&& self.binary_modifier_preserves_tsid_join_key(left_context, right_context, modifier)
|
||||
&& left_context.use_tsid
|
||||
&& right_context.use_tsid
|
||||
&& has_tsid(left_schema)
|
||||
&& has_tsid(right_schema);
|
||||
|
||||
let (mut left_tag_columns, mut right_tag_columns) = if use_tsid_join {
|
||||
(
|
||||
@@ -3999,25 +4029,22 @@ impl PromPlanner {
|
||||
BTreeSet::from([DATA_SCHEMA_TSID_COLUMN_NAME.to_string()]),
|
||||
)
|
||||
} else {
|
||||
let left_tag_columns = if only_join_time_index {
|
||||
BTreeSet::new()
|
||||
if only_join_time_index {
|
||||
(BTreeSet::new(), BTreeSet::new())
|
||||
} else {
|
||||
left_ctx
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>()
|
||||
};
|
||||
let right_tag_columns = if only_join_time_index {
|
||||
BTreeSet::new()
|
||||
} else {
|
||||
right_ctx
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>()
|
||||
};
|
||||
(left_tag_columns, right_tag_columns)
|
||||
(
|
||||
left_context
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>(),
|
||||
right_context
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>(),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
if !use_tsid_join
|
||||
@@ -4039,7 +4066,56 @@ impl PromPlanner {
|
||||
}
|
||||
}
|
||||
|
||||
(left_tag_columns, right_tag_columns)
|
||||
let force_empty_join =
|
||||
!use_tsid_join && !only_join_time_index && left_tag_columns != right_tag_columns;
|
||||
if force_empty_join {
|
||||
let common_tag_columns = left_tag_columns
|
||||
.intersection(&right_tag_columns)
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>();
|
||||
left_tag_columns = common_tag_columns.clone();
|
||||
right_tag_columns = common_tag_columns;
|
||||
}
|
||||
|
||||
Ok((left_tag_columns, right_tag_columns, force_empty_join))
|
||||
}
|
||||
|
||||
fn binary_modifier_preserves_tsid_join_key(
|
||||
&self,
|
||||
left_context: &PromPlannerContext,
|
||||
right_context: &PromPlannerContext,
|
||||
modifier: &Option<BinModifier>,
|
||||
) -> bool {
|
||||
let Some(modifier) = modifier else {
|
||||
return true;
|
||||
};
|
||||
|
||||
if !matches!(modifier.card, VectorMatchCardinality::OneToOne) {
|
||||
return false;
|
||||
}
|
||||
|
||||
match &modifier.matching {
|
||||
None => true,
|
||||
Some(LabelModifier::Exclude(ignoring)) => ignoring.labels.iter().all(|label| {
|
||||
!left_context.tag_columns.contains(label)
|
||||
&& !right_context.tag_columns.contains(label)
|
||||
}),
|
||||
Some(LabelModifier::Include(on)) => {
|
||||
let on_labels = on.labels.iter().cloned().collect::<BTreeSet<_>>();
|
||||
let left_labels = left_context
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>();
|
||||
let right_labels = right_context
|
||||
.tag_columns
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<BTreeSet<_>>();
|
||||
|
||||
on_labels == left_labels && on_labels == right_labels
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a inner join on time index column and tag columns to concat two logical plans.
|
||||
@@ -4055,15 +4131,18 @@ impl PromPlanner {
|
||||
right_time_index_column: Option<String>,
|
||||
only_join_time_index: bool,
|
||||
modifier: &Option<BinModifier>,
|
||||
left_context: &PromPlannerContext,
|
||||
right_context: &PromPlannerContext,
|
||||
) -> Result<LogicalPlan> {
|
||||
let (mut left_tag_columns, mut right_tag_columns) = Self::binary_join_key_columns(
|
||||
&left,
|
||||
&right,
|
||||
&self.ctx,
|
||||
&self.ctx,
|
||||
only_join_time_index,
|
||||
modifier,
|
||||
);
|
||||
let (mut left_tag_columns, mut right_tag_columns, force_empty_join) = self
|
||||
.binary_join_key_columns(
|
||||
left.schema(),
|
||||
right.schema(),
|
||||
left_context,
|
||||
right_context,
|
||||
only_join_time_index,
|
||||
modifier,
|
||||
)?;
|
||||
|
||||
// push time index column if it exists
|
||||
if let (Some(left_time_index_column), Some(right_time_index_column)) =
|
||||
@@ -4096,7 +4175,7 @@ impl PromPlanner {
|
||||
.map(Column::from_name)
|
||||
.collect::<Vec<_>>(),
|
||||
),
|
||||
None,
|
||||
force_empty_join.then_some(lit(false)),
|
||||
NullEquality::NullEqualsNull,
|
||||
)
|
||||
.context(DataFusionPlanningSnafu)?
|
||||
@@ -4859,14 +4938,29 @@ mod test {
|
||||
async fn build_test_table_provider_with_tsid_fields(
|
||||
table_specs: &[((String, String), usize)],
|
||||
num_tag: usize,
|
||||
) -> DfTableSourceProvider {
|
||||
let table_specs = table_specs
|
||||
.iter()
|
||||
.map(|(table_name_tuple, num_field)| (table_name_tuple.clone(), num_tag, *num_field))
|
||||
.collect::<Vec<_>>();
|
||||
build_test_table_provider_with_tsid_tag_fields(&table_specs).await
|
||||
}
|
||||
|
||||
async fn build_test_table_provider_with_tsid_tag_fields(
|
||||
table_specs: &[((String, String), usize, usize)],
|
||||
) -> DfTableSourceProvider {
|
||||
let catalog_list = MemoryCatalogManager::with_default_setup();
|
||||
|
||||
let physical_table_name = "phy";
|
||||
let physical_table_id = 999u32;
|
||||
let physical_num_tag = table_specs
|
||||
.iter()
|
||||
.map(|(_, num_tag, _)| *num_tag)
|
||||
.max()
|
||||
.unwrap_or(0);
|
||||
let physical_num_field = table_specs
|
||||
.iter()
|
||||
.map(|(_, num_field)| *num_field)
|
||||
.map(|(_, _, num_field)| *num_field)
|
||||
.max()
|
||||
.unwrap_or(0);
|
||||
|
||||
@@ -4884,7 +4978,7 @@ mod test {
|
||||
false,
|
||||
),
|
||||
];
|
||||
for i in 0..num_tag {
|
||||
for i in 0..physical_num_tag {
|
||||
columns.push(ColumnSchema::new(
|
||||
format!("tag_{i}"),
|
||||
ConcreteDataType::string_datatype(),
|
||||
@@ -4908,11 +5002,13 @@ mod test {
|
||||
}
|
||||
|
||||
let schema = Arc::new(Schema::new(columns));
|
||||
let primary_key_indices = (0..(2 + num_tag)).collect::<Vec<_>>();
|
||||
let primary_key_indices = (0..(2 + physical_num_tag)).collect::<Vec<_>>();
|
||||
let table_meta = TableMetaBuilder::empty()
|
||||
.schema(schema)
|
||||
.primary_key_indices(primary_key_indices)
|
||||
.value_indices((2 + num_tag..2 + num_tag + 1 + physical_num_field).collect())
|
||||
.value_indices(
|
||||
(2 + physical_num_tag..2 + physical_num_tag + 1 + physical_num_field).collect(),
|
||||
)
|
||||
.engine(METRIC_ENGINE_NAME.to_string())
|
||||
.next_column_id(1024)
|
||||
.build()
|
||||
@@ -4939,9 +5035,10 @@ mod test {
|
||||
}
|
||||
|
||||
// Register metric engine logical tables without `__tsid`, referencing the physical table.
|
||||
for (idx, ((schema_name, table_name), num_field)) in table_specs.iter().enumerate() {
|
||||
for (idx, ((schema_name, table_name), num_tag, num_field)) in table_specs.iter().enumerate()
|
||||
{
|
||||
let mut columns = vec![];
|
||||
for i in 0..num_tag {
|
||||
for i in 0..*num_tag {
|
||||
columns.push(ColumnSchema::new(
|
||||
format!("tag_{i}"),
|
||||
ConcreteDataType::string_datatype(),
|
||||
@@ -4973,8 +5070,8 @@ mod test {
|
||||
let table_id = 1024u32 + idx as u32;
|
||||
let table_meta = TableMetaBuilder::empty()
|
||||
.schema(schema)
|
||||
.primary_key_indices((0..num_tag).collect())
|
||||
.value_indices((num_tag + 1..num_tag + 1 + *num_field).collect())
|
||||
.primary_key_indices((0..*num_tag).collect())
|
||||
.value_indices((*num_tag + 1..*num_tag + 1 + *num_field).collect())
|
||||
.engine(METRIC_ENGINE_NAME.to_string())
|
||||
.options(options)
|
||||
.next_column_id(1024)
|
||||
@@ -5443,6 +5540,71 @@ mod test {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timestamp_binary_join_falls_back_when_tsid_is_projected_out() {
|
||||
for query in [
|
||||
"timestamp(some_metric) / some_metric",
|
||||
"some_metric / timestamp(some_metric)",
|
||||
] {
|
||||
let eval_stmt = build_eval_stmt(query);
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid(
|
||||
&[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())],
|
||||
1,
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(!plan_str.contains("__tsid ="), "{query}: {plan_str}");
|
||||
assert!(
|
||||
plan_str.contains("lhs.tag_0 = rhs.tag_0"),
|
||||
"{query}: {plan_str}"
|
||||
);
|
||||
assert!(
|
||||
!plan
|
||||
.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.any(|field| field.name() == DATA_SCHEMA_TSID_COLUMN_NAME),
|
||||
"{query}: {plan_str}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timestamp_binary_join_rejects_default_matching_on_mismatched_labels() {
|
||||
let eval_stmt = build_eval_stmt("timestamp(left_host_job) / right_by_job");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid_tag_fields(&[
|
||||
(
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "left_host_job".to_string()),
|
||||
2,
|
||||
1,
|
||||
),
|
||||
(
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "right_by_job".to_string()),
|
||||
1,
|
||||
1,
|
||||
),
|
||||
])
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
|
||||
assert!(
|
||||
plan_str.contains("Boolean(false)") || plan_str.contains("false"),
|
||||
"{plan_str}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tsid_is_preserved_for_nested_default_binary_joins() {
|
||||
let eval_stmt = build_eval_stmt("(some_metric - some_alt_metric) / some_third_metric");
|
||||
@@ -5866,6 +6028,51 @@ mod test {
|
||||
assert_eq!(value_columns, 1, "{field_names:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn comparison_binary_join_uses_shorter_field_side() {
|
||||
let eval_stmt = build_eval_stmt("two_field_metric > one_field_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid_fields(
|
||||
&[
|
||||
(
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"two_field_metric".to_string(),
|
||||
),
|
||||
2,
|
||||
),
|
||||
(
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"one_field_metric".to_string(),
|
||||
),
|
||||
1,
|
||||
),
|
||||
],
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let field_names = plan
|
||||
.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.map(|field| field.name().clone())
|
||||
.collect::<Vec<_>>();
|
||||
assert!(
|
||||
field_names.iter().any(|name| name == "field_0"),
|
||||
"{field_names:?}"
|
||||
);
|
||||
assert!(
|
||||
!field_names.iter().any(|name| name == "field_1"),
|
||||
"{field_names:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn label_matching_modifier_disables_tsid_binary_join() {
|
||||
let eval_stmt = build_eval_stmt("some_metric / ignoring(tag_0) some_alt_metric");
|
||||
@@ -5895,6 +6102,161 @@ mod test {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ignoring_absent_label_keeps_tsid_binary_join() {
|
||||
let eval_stmt = build_eval_stmt("some_metric / ignoring(missing) some_alt_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid(
|
||||
&[
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()),
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"some_alt_metric".to_string(),
|
||||
),
|
||||
],
|
||||
2,
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(
|
||||
plan_str.contains("some_metric.__tsid = some_alt_metric.__tsid"),
|
||||
"{plan_str}"
|
||||
);
|
||||
assert!(!plan_str.contains("tag_0 ="), "{plan_str}");
|
||||
assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn range_function_keeps_tsid_for_absent_ignoring_binary_join() {
|
||||
let eval_stmt =
|
||||
build_eval_stmt("rate(some_metric[5m]) / ignoring(missing) some_alt_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid(
|
||||
&[
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()),
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"some_alt_metric".to_string(),
|
||||
),
|
||||
],
|
||||
2,
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(
|
||||
plan_str.contains("some_metric.__tsid = some_alt_metric.__tsid"),
|
||||
"{plan_str}"
|
||||
);
|
||||
assert!(!plan_str.contains("tag_0 ="), "{plan_str}");
|
||||
assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn on_full_label_set_keeps_tsid_binary_join() {
|
||||
let eval_stmt = build_eval_stmt("some_metric / on(tag_0, tag_1) some_alt_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid(
|
||||
&[
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()),
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"some_alt_metric".to_string(),
|
||||
),
|
||||
],
|
||||
2,
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(
|
||||
plan_str.contains("some_metric.__tsid = some_alt_metric.__tsid"),
|
||||
"{plan_str}"
|
||||
);
|
||||
assert!(!plan_str.contains("tag_0 ="), "{plan_str}");
|
||||
assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn on_partial_label_set_disables_tsid_binary_join() {
|
||||
let eval_stmt = build_eval_stmt("some_metric / on(tag_0) some_alt_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid(
|
||||
&[
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()),
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"some_alt_metric".to_string(),
|
||||
),
|
||||
],
|
||||
2,
|
||||
1,
|
||||
)
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(!plan_str.contains("__tsid ="), "{plan_str}");
|
||||
assert!(
|
||||
plan_str.contains("some_metric.tag_0 = some_alt_metric.tag_0"),
|
||||
"{plan_str}"
|
||||
);
|
||||
assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn on_label_set_must_cover_both_sides_to_use_tsid_binary_join() {
|
||||
let eval_stmt = build_eval_stmt("some_metric / on(tag_0) some_alt_metric");
|
||||
|
||||
let table_provider = build_test_table_provider_with_tsid_tag_fields(&[
|
||||
(
|
||||
(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string()),
|
||||
2,
|
||||
1,
|
||||
),
|
||||
(
|
||||
(
|
||||
DEFAULT_SCHEMA_NAME.to_string(),
|
||||
"some_alt_metric".to_string(),
|
||||
),
|
||||
1,
|
||||
1,
|
||||
),
|
||||
])
|
||||
.await;
|
||||
let plan =
|
||||
PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let plan_str = plan.display_indent_schema().to_string();
|
||||
assert!(!plan_str.contains("__tsid ="), "{plan_str}");
|
||||
assert!(
|
||||
plan_str.contains("some_metric.tag_0 = some_alt_metric.tag_0"),
|
||||
"{plan_str}"
|
||||
);
|
||||
assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn comparison_binary_join_uses_tsid_and_keeps_it_in_filtered_result() {
|
||||
let eval_stmt = build_eval_stmt("some_metric > some_alt_metric");
|
||||
|
||||
@@ -100,31 +100,31 @@ TQL EXPLAIN (
|
||||
)
|
||||
);
|
||||
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
| plan_type | plan |
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
| logical_plan | HistogramFold: le=le, field=sum(prom_avg_over_time(ts_range,v)), quantile=0.5 |
|
||||
| | MergeScan [is_placeholder=false, remote_input=[ |
|
||||
| | Sort: test_tsid.le ASC NULLS LAST, test_tsid.tag4 ASC NULLS LAST, test_tsid.tag5 ASC NULLS LAST, test_tsid.ts ASC NULLS LAST |
|
||||
| | Aggregate: groupBy=[[test_tsid.le, test_tsid.tag4, test_tsid.tag5, test_tsid.ts]], aggr=[[sum(prom_avg_over_time(ts_range,v))]] |
|
||||
| | Filter: prom_avg_over_time(ts_range,v) IS NOT NULL |
|
||||
| | Projection: test_tsid.ts, prom_avg_over_time(ts_range, v) AS prom_avg_over_time(ts_range,v), test_tsid.le, test_tsid.tag1, test_tsid.tag2, test_tsid.tag4, test_tsid.tag5, test_tsid.tag6, test_tsid.tag7, test_tsid.tag8 |
|
||||
| | PromRangeManipulate: req range=[1769139000000..1769139900000], interval=[60000], eval range=[1800000], time index=[ts], values=["v"] |
|
||||
| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] |
|
||||
| | PromSeriesDivide: tags=["__tsid"] |
|
||||
| | Sort: test_tsid.__tsid ASC NULLS FIRST, test_tsid.ts ASC NULLS FIRST |
|
||||
| | Filter: test_tsid.ts >= TimestampMillisecond(1769137200001, None) AND test_tsid.ts <= TimestampMillisecond(1769139900000, None) |
|
||||
| | Projection: test_tsid.v, test_tsid.le, test_tsid.tag1, test_tsid.tag2, test_tsid.tag4, test_tsid.tag5, test_tsid.tag6, test_tsid.tag7, test_tsid.tag8, test_tsid.__tsid, test_tsid.ts |
|
||||
| | SubqueryAlias: test_tsid |
|
||||
| | Filter: phy.__table_id=UInt32(REDACTED) |
|
||||
| | TableScan: phy projection=[ts, v, tag1, tag2, le, tag4, tag5, tag6, tag7, tag8, __table_id, __tsid] |
|
||||
| | ]] |
|
||||
| physical_plan | HistogramFoldExec: le=@0, field=@4, quantile=0.5 |
|
||||
| | SortExec: expr=[tag4@1 ASC NULLS LAST, tag5@2 ASC NULLS LAST, ts@3 ASC NULLS LAST, CAST(le@0 AS Float64) ASC NULLS LAST], preserve_partitioning=[true] |
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
| plan_type | plan |
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
| logical_plan | HistogramFold: le=le, field=sum(prom_avg_over_time(ts_range,v)), quantile=0.5 |
|
||||
| | MergeScan [is_placeholder=false, remote_input=[ |
|
||||
| | Sort: test_tsid.le ASC NULLS LAST, test_tsid.tag4 ASC NULLS LAST, test_tsid.tag5 ASC NULLS LAST, test_tsid.ts ASC NULLS LAST |
|
||||
| | Aggregate: groupBy=[[test_tsid.le, test_tsid.tag4, test_tsid.tag5, test_tsid.ts]], aggr=[[sum(prom_avg_over_time(ts_range,v))]] |
|
||||
| | Filter: prom_avg_over_time(ts_range,v) IS NOT NULL |
|
||||
| | Projection: test_tsid.ts, prom_avg_over_time(ts_range, v) AS prom_avg_over_time(ts_range,v), test_tsid.le, test_tsid.tag1, test_tsid.tag2, test_tsid.tag4, test_tsid.tag5, test_tsid.tag6, test_tsid.tag7, test_tsid.tag8, test_tsid.__tsid |
|
||||
| | PromRangeManipulate: req range=[1769139000000..1769139900000], interval=[60000], eval range=[1800000], time index=[ts], values=["v"] |
|
||||
| | PromSeriesNormalize: offset=[0], time index=[ts], filter NaN: [true] |
|
||||
| | PromSeriesDivide: tags=["__tsid"] |
|
||||
| | Sort: test_tsid.__tsid ASC NULLS FIRST, test_tsid.ts ASC NULLS FIRST |
|
||||
| | Filter: test_tsid.ts >= TimestampMillisecond(1769137200001, None) AND test_tsid.ts <= TimestampMillisecond(1769139900000, None) |
|
||||
| | Projection: test_tsid.v, test_tsid.le, test_tsid.tag1, test_tsid.tag2, test_tsid.tag4, test_tsid.tag5, test_tsid.tag6, test_tsid.tag7, test_tsid.tag8, test_tsid.__tsid, test_tsid.ts |
|
||||
| | SubqueryAlias: test_tsid |
|
||||
| | Filter: phy.__table_id=UInt32(REDACTED) |
|
||||
| | TableScan: phy projection=[ts, v, tag1, tag2, le, tag4, tag5, tag6, tag7, tag8, __table_id, __tsid] |
|
||||
| | ]] |
|
||||
| physical_plan | HistogramFoldExec: le=@0, field=@4, quantile=0.5 |
|
||||
| | SortExec: expr=[tag4@1 ASC NULLS LAST, tag5@2 ASC NULLS LAST, ts@3 ASC NULLS LAST, CAST(le@0 AS Float64) ASC NULLS LAST], preserve_partitioning=[true] |
|
||||
| | RepartitionExec: REDACTED
|
||||
| | MergeScanExec: REDACTED
|
||||
| | |
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
| | |
|
||||
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|
||||
|
||||
CREATE FLOW IF NOT EXISTS test_tsid
|
||||
SINK TO 'test_tsid_output'
|
||||
|
||||
@@ -202,38 +202,14 @@ TQL EVAL(0, 15, '5s') label_replace(vector(1), "host", "host1", "", "");
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1), "host", "host1", "", "");
|
||||
|
||||
+-------+---------------------+----------------------------+
|
||||
| host | time | test.val * .greptime_value |
|
||||
+-------+---------------------+----------------------------+
|
||||
| host1 | 1970-01-01T00:00:00 | 1.0 |
|
||||
| host1 | 1970-01-01T00:00:05 | 1.0 |
|
||||
| host1 | 1970-01-01T00:00:05 | 3.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 1.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 3.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 5.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 1.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 3.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 5.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 7.0 |
|
||||
+-------+---------------------+----------------------------+
|
||||
++
|
||||
++
|
||||
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} + label_replace(vector(1), "host", "host1", "", "");
|
||||
|
||||
+-------+---------------------+----------------------------+
|
||||
| host | time | test.val + .greptime_value |
|
||||
+-------+---------------------+----------------------------+
|
||||
| host1 | 1970-01-01T00:00:00 | 2.0 |
|
||||
| host1 | 1970-01-01T00:00:05 | 2.0 |
|
||||
| host1 | 1970-01-01T00:00:05 | 4.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 2.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 4.0 |
|
||||
| host1 | 1970-01-01T00:00:10 | 6.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 2.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 4.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 6.0 |
|
||||
| host1 | 1970-01-01T00:00:15 | 8.0 |
|
||||
+-------+---------------------+----------------------------+
|
||||
++
|
||||
++
|
||||
|
||||
-- Empty regex with existing source label
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
@@ -303,7 +279,8 @@ TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1),
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL(0, 15, '5s') {__name__="test",host="host1"} * label_replace(vector(1), "addr", "host1", "instance", "");
|
||||
|
||||
Error: 3000(PlanQuery), Internal error during building DataFusion plan: No field named addr. Valid fields are test.ts, test.host, test.idc, test.val.
|
||||
++
|
||||
++
|
||||
|
||||
TQL EVAL label_replace(demo_num_cpus, "~invalid", "", "src", "(.*)");
|
||||
|
||||
|
||||
@@ -38,6 +38,20 @@ WITH(
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
CREATE TABLE tsid_binary_join_right_by_job (
|
||||
job STRING NULL,
|
||||
ts TIMESTAMP(3) NOT NULL,
|
||||
greptime_value DOUBLE NULL,
|
||||
TIME INDEX (ts),
|
||||
PRIMARY KEY(job),
|
||||
)
|
||||
ENGINE = metric
|
||||
WITH(
|
||||
on_physical_table = 'tsid_binary_join_physical'
|
||||
);
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
CREATE TABLE tsid_binary_join_third (
|
||||
host STRING NULL,
|
||||
job STRING NULL,
|
||||
@@ -69,6 +83,14 @@ INSERT INTO tsid_binary_join_right (host, job, ts, greptime_value) VALUES
|
||||
|
||||
Affected Rows: 4
|
||||
|
||||
INSERT INTO tsid_binary_join_right_by_job (job, ts, greptime_value) VALUES
|
||||
('job1', 0, 3),
|
||||
('job2', 0, 6),
|
||||
('job1', 5000, 5),
|
||||
('job2', 5000, 7);
|
||||
|
||||
Affected Rows: 4
|
||||
|
||||
INSERT INTO tsid_binary_join_third (host, job, ts, greptime_value) VALUES
|
||||
('host1', 'job1', 0, 2),
|
||||
('host2', 'job2', 0, 3),
|
||||
@@ -244,6 +266,45 @@ TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / ignoring(host) tsid_binary_join
|
||||
|_|_| Total rows: 4_|
|
||||
+-+-+-+
|
||||
|
||||
-- `on(job)` must stay label-based when only the left side has extra row-key labels.
|
||||
-- SQLNESS REPLACE (metrics.*) REDACTED
|
||||
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
|
||||
-- SQLNESS REPLACE (-+) -
|
||||
-- SQLNESS REPLACE (\s\s+) _
|
||||
-- SQLNESS REPLACE (peers.*) REDACTED
|
||||
-- SQLNESS REPLACE Hash\(\[job@1,\sts@2\],.* Hash([job@1, ts@2],REDACTED
|
||||
-- SQLNESS REPLACE Hash\(\[job@1,\sts@3\],.* Hash([job@1, ts@3],REDACTED
|
||||
-- SQLNESS REPLACE input_partitions=\d+ input_partitions=REDACTED
|
||||
-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED
|
||||
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
|
||||
TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / on(job) tsid_binary_join_right_by_job;
|
||||
|
||||
+-+-+-+
|
||||
| stage | node | plan_|
|
||||
+-+-+-+
|
||||
| 0_| 0_|_ProjectionExec: expr=[job@2 as job, ts@4 as ts, __tsid@3 as __tsid, greptime_value@0 / greptime_value@1 as tsid_binary_join_left.greptime_value / tsid_binary_join_right_by_job.greptime_value] REDACTED
|
||||
|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(job@1, job@1), (ts@2, ts@3)], projection=[greptime_value@0, greptime_value@3, job@4, __tsid@5, ts@6], NullsEqual: true REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([job@1, ts@2],REDACTED
|
||||
|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, ts@4 as ts] REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([job@1, ts@3],REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_|
|
||||
| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED
|
||||
|_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED
|
||||
|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, host@3 as host, job@4 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED
|
||||
|_|_|_CooperativeExec REDACTED
|
||||
|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries" REDACTED
|
||||
|_|_|_|
|
||||
| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED
|
||||
|_|_|_PromSeriesDivideExec: tags=["__tsid"] REDACTED
|
||||
|_|_|_ProjectionExec: expr=[greptime_value@1 as greptime_value, job@3 as job, __tsid@2 as __tsid, ts@0 as ts] REDACTED
|
||||
|_|_|_CooperativeExec REDACTED
|
||||
|_|_|_SeriesScan: region=REDACTED, "partition_count":REDACTED, "distribution":"PerSeries" REDACTED
|
||||
|_|_|_|
|
||||
|_|_| Total rows: 4_|
|
||||
+-+-+-+
|
||||
|
||||
-- Comparison filters can join on `__tsid`, but the filtered result must still behave like
|
||||
-- a regular derived vector downstream.
|
||||
-- SQLNESS REPLACE (metrics.*) REDACTED
|
||||
@@ -390,7 +451,8 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > bool tsid_binary_join_right)
|
||||
| stage | node | plan_|
|
||||
+-+-+-+
|
||||
| 0_| 0_|_ProjectionExec: expr=[host@2 as host, job@3 as job, ts@5 as ts, __tsid@4 as __tsid, tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value@0 / greptime_value@1 as lhs.tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value / rhs.greptime_value] REDACTED
|
||||
|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@0, ts@4)], projection=[tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value@2, greptime_value@3, host@4, job@5, __tsid@6, ts@7], NullsEqual: true REDACTED
|
||||
|_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@0, ts@4)], projection=[tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value@2, greptime_value@3, host@4, job@5, __tsid@6, ts@7], NullsEqual: true REDACTED
|
||||
|_|_|_CoalescePartitionsExec REDACTED
|
||||
|_|_|_ProjectionExec: expr=[ts@3 as ts, __tsid@2 as __tsid, tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value@0 + greptime_value@1 as tsid_binary_join_right.tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value + tsid_binary_join_left.greptime_value] REDACTED
|
||||
|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@1), (ts@0, ts@2)], projection=[tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value@2, greptime_value@3, __tsid@4, ts@5], NullsEqual: true REDACTED
|
||||
|_|_|_ProjectionExec: expr=[ts@3 as ts, __tsid@2 as __tsid, CAST(greptime_value@1 < greptime_value@0 AS Float64) as tsid_binary_join_left.greptime_value > tsid_binary_join_right.greptime_value] REDACTED
|
||||
@@ -404,7 +466,7 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left > bool tsid_binary_join_right)
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, __tsid@3 as __tsid, ts@4 as ts] REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
|_|_|_CooperativeExec REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_|
|
||||
| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED
|
||||
@@ -501,8 +563,8 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left / ignoring(host) group_left tsid
|
||||
| stage | node | plan_|
|
||||
+-+-+-+
|
||||
| 0_| 0_|_ProjectionExec: expr=[host@2 as host, job@3 as job, ts@5 as ts, __tsid@4 as __tsid, tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value@0 / greptime_value@1 as tsid_binary_join_right.tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value / tsid_binary_join_left.greptime_value] REDACTED
|
||||
|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@0, ts@4)], projection=[tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value@2, greptime_value@3, host@4, job@5, __tsid@6, ts@7], NullsEqual: true REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
|_|_|_HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__tsid@1, __tsid@3), (ts@0, ts@4)], projection=[tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value@2, greptime_value@3, host@4, job@5, __tsid@6, ts@7], NullsEqual: true REDACTED
|
||||
|_|_|_CoalescePartitionsExec REDACTED
|
||||
|_|_|_ProjectionExec: expr=[ts@3 as ts, __tsid@2 as __tsid, greptime_value@0 / greptime_value@1 as tsid_binary_join_left.greptime_value / tsid_binary_join_right.greptime_value] REDACTED
|
||||
|_|_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(job@1, job@1), (ts@2, ts@3)], projection=[greptime_value@0, greptime_value@3, __tsid@5, ts@6], NullsEqual: true REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
@@ -511,7 +573,7 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left / ignoring(host) group_left tsid
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
|_|_|_ProjectionExec: expr=[greptime_value@0 as greptime_value, job@2 as job, __tsid@3 as __tsid, ts@4 as ts] REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_RepartitionExec: partitioning=Hash([REDACTED
|
||||
|_|_|_CooperativeExec REDACTED
|
||||
|_|_|_MergeScanExec: REDACTED
|
||||
|_|_|_|
|
||||
| 1_| 0_|_PromInstantManipulateExec: range=[0..5000], lookback=[300000], interval=[5000], time index=[ts] REDACTED
|
||||
@@ -547,6 +609,18 @@ TQL EVAL (0, 5, '5s') tsid_binary_join_left / tsid_binary_join_right;
|
||||
| host2 | job2 | 1970-01-01T00:00:05 | 3.0 |
|
||||
+-------+------+---------------------+------------------------------------------------------------------------------+
|
||||
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL (0, 5, '5s') tsid_binary_join_left / on(job) tsid_binary_join_right_by_job;
|
||||
|
||||
+------+---------------------+-------------------------------------------------------------------------------------+
|
||||
| job | ts | tsid_binary_join_left.greptime_value / tsid_binary_join_right_by_job.greptime_value |
|
||||
+------+---------------------+-------------------------------------------------------------------------------------+
|
||||
| job1 | 1970-01-01T00:00:00 | 4.0 |
|
||||
| job1 | 1970-01-01T00:00:05 | 3.0 |
|
||||
| job2 | 1970-01-01T00:00:00 | 3.0 |
|
||||
| job2 | 1970-01-01T00:00:05 | 3.0 |
|
||||
+------+---------------------+-------------------------------------------------------------------------------------+
|
||||
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL (0, 5, '5s') (tsid_binary_join_left + tsid_binary_join_right) / tsid_binary_join_left;
|
||||
|
||||
@@ -621,6 +695,10 @@ DROP TABLE tsid_binary_join_third;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
DROP TABLE tsid_binary_join_right_by_job;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
DROP TABLE tsid_binary_join_right;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
@@ -33,6 +33,18 @@ WITH(
|
||||
on_physical_table = 'tsid_binary_join_physical'
|
||||
);
|
||||
|
||||
CREATE TABLE tsid_binary_join_right_by_job (
|
||||
job STRING NULL,
|
||||
ts TIMESTAMP(3) NOT NULL,
|
||||
greptime_value DOUBLE NULL,
|
||||
TIME INDEX (ts),
|
||||
PRIMARY KEY(job),
|
||||
)
|
||||
ENGINE = metric
|
||||
WITH(
|
||||
on_physical_table = 'tsid_binary_join_physical'
|
||||
);
|
||||
|
||||
CREATE TABLE tsid_binary_join_third (
|
||||
host STRING NULL,
|
||||
job STRING NULL,
|
||||
@@ -58,6 +70,12 @@ INSERT INTO tsid_binary_join_right (host, job, ts, greptime_value) VALUES
|
||||
('host1', 'job1', 5000, 5),
|
||||
('host2', 'job2', 5000, 7);
|
||||
|
||||
INSERT INTO tsid_binary_join_right_by_job (job, ts, greptime_value) VALUES
|
||||
('job1', 0, 3),
|
||||
('job2', 0, 6),
|
||||
('job1', 5000, 5),
|
||||
('job2', 5000, 7);
|
||||
|
||||
INSERT INTO tsid_binary_join_third (host, job, ts, greptime_value) VALUES
|
||||
('host1', 'job1', 0, 2),
|
||||
('host2', 'job2', 0, 3),
|
||||
@@ -117,6 +135,19 @@ TQL ANALYZE (0, 5, '5s') ((tsid_binary_join_left + tsid_binary_join_right) * (ts
|
||||
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
|
||||
TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / ignoring(host) tsid_binary_join_right;
|
||||
|
||||
-- `on(job)` must stay label-based when only the left side has extra row-key labels.
|
||||
-- SQLNESS REPLACE (metrics.*) REDACTED
|
||||
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
|
||||
-- SQLNESS REPLACE (-+) -
|
||||
-- SQLNESS REPLACE (\s\s+) _
|
||||
-- SQLNESS REPLACE (peers.*) REDACTED
|
||||
-- SQLNESS REPLACE Hash\(\[job@1,\sts@2\],.* Hash([job@1, ts@2],REDACTED
|
||||
-- SQLNESS REPLACE Hash\(\[job@1,\sts@3\],.* Hash([job@1, ts@3],REDACTED
|
||||
-- SQLNESS REPLACE input_partitions=\d+ input_partitions=REDACTED
|
||||
-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED
|
||||
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
|
||||
TQL ANALYZE (0, 5, '5s') tsid_binary_join_left / on(job) tsid_binary_join_right_by_job;
|
||||
|
||||
-- Comparison filters can join on `__tsid`, but the filtered result must still behave like
|
||||
-- a regular derived vector downstream.
|
||||
-- SQLNESS REPLACE (metrics.*) REDACTED
|
||||
@@ -198,6 +229,9 @@ TQL ANALYZE (0, 5, '5s') (tsid_binary_join_left / ignoring(host) group_left tsid
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL (0, 5, '5s') tsid_binary_join_left / tsid_binary_join_right;
|
||||
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL (0, 5, '5s') tsid_binary_join_left / on(job) tsid_binary_join_right_by_job;
|
||||
|
||||
-- SQLNESS SORT_RESULT 3 1
|
||||
TQL EVAL (0, 5, '5s') (tsid_binary_join_left + tsid_binary_join_right) / tsid_binary_join_left;
|
||||
|
||||
@@ -219,6 +253,7 @@ TQL EVAL (0, 5, '5s') (tsid_binary_join_left or tsid_binary_join_right) / tsid_b
|
||||
TQL EVAL (0, 5, '5s') rate(tsid_binary_join_left[5s]) / tsid_binary_join_left;
|
||||
|
||||
DROP TABLE tsid_binary_join_third;
|
||||
DROP TABLE tsid_binary_join_right_by_job;
|
||||
DROP TABLE tsid_binary_join_right;
|
||||
DROP TABLE tsid_binary_join_left;
|
||||
DROP TABLE tsid_binary_join_physical;
|
||||
|
||||
Reference in New Issue
Block a user