fix(promql): resolve the OR right operand's time-index column from its schema (#10335)

`or` inferred the qualifier of an operand plan from its first field
(`schema.qualified_field(0)`). A binary operation with an explicit matching
modifier derives its result labels from the left operand when the match
reduces the label set, so its projected plan carries two qualifiers: the
labels are `lhs.<label>` while the time index keeps the context table's
qualifier (`rhs.<greptime_timestamp>`). Prefixing the time index with the
first field's qualifier then referenced a column that does not exist, and
planning failed with

  Internal error during building DataFusion plan: No field named
  lhs.greptime_timestamp. Did you mean 'rhs.greptime_timestamp'?

for e.g. `sum by (l) (f(a) / ignoring(x) group_left() f(b))` used as the
right operand of `or on (l)`. `f(a) / ignoring(x) f(b)` triggers it as
well, so this is not specific to a group modifier. Without a matching
modifier every projected column takes the right operand's qualifier, which
is why plain `a / b` never hit this.

Resolve the time index against the actual operand schema: keep the
preferred qualifier when it really holds the column (uniform schemas, so no
plan changes), otherwise take the qualifier of the column found by name and
let DataFusion report ambiguity instead of guessing.

Signed-off-by: discord9 <discord9@163.com>

src/query/src/promql/planner/set_operator.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use datafusion::logical_expr::{Cast, Extension, LogicalPlan, LogicalPlanBuilder};
 use datafusion::prelude::{Column, Expr as DfExpr, JoinType};
 use datafusion::scalar::ScalarValue;
-use datafusion_common::{NullEquality, TableReference};
+use datafusion_common::{DFSchema, NullEquality, TableReference};
 use datatypes::arrow::datatypes::DataType as ArrowDataType;
 use promql::extension_plan::UnionDistinctOn;
 use promql_parser::parser::token::{self, TokenType};
@@ -404,10 +404,17 @@ impl PromPlanner {
                 DfExpr::Column(Column::new(None::<String>, col))
             }
         });
-        let right_time_index_expr = DfExpr::Column(Column::new(
+        // The right operand does not necessarily qualify every column with the qualifier of its
+        // first field: a binary operation with a matching modifier (`on`, `ignoring`, `group_left`,
+        // `group_right`) derives its result labels from the left operand and projects them under
+        // that side's qualifier (`lhs.<label>`), while its time index keeps the qualifier of the
+        // context's table. Resolve the time index against the actual schema so the reference
+        // cannot miss.
+        let right_time_index_expr = DfExpr::Column(Self::resolved_column(
+            right.schema(),
             right_qualifier.clone(),
-            right_time_index_column,
-        ))
+            &right_time_index_column,
+        )?)
         .alias(left_time_index_column.clone());
         // The field column in right side may not have qualifier (it may be removed by join operation),
         // so we need to find it from the schema.
@@ -633,6 +640,32 @@ impl PromPlanner {
         Ok(result)
     }
 
+    /// Resolves the column `name` of `schema`, preferring `preferred_qualifier`.
+    ///
+    /// A single operand plan can mix qualifiers, so the qualifier of its first field is not a
+    /// reliable prefix for every column of that plan. When `preferred_qualifier` does not hold the
+    /// column, the fallback is `qualified_field_with_unqualified_name`: it resolves a unique name
+    /// match, or a single unqualified candidate next to qualified ones, and reports
+    /// `AmbiguousReference` otherwise. This mirrors how the planner already resolves derived
+    /// columns against a join output.
+    fn resolved_column(
+        schema: &DFSchema,
+        preferred_qualifier: Option<TableReference>,
+        name: &str,
+    ) -> Result<Column> {
+        let qualifier = match &preferred_qualifier {
+            Some(qualifier) if schema.field_with_name(Some(qualifier), name).is_ok() => {
+                preferred_qualifier.clone()
+            }
+            _ => schema
+                .qualified_field_with_unqualified_name(name)
+                .context(DataFusionPlanningSnafu)?
+                .0
+                .cloned(),
+        };
+        Ok(Column::new(qualifier, name))
+    }
+
     /// Build a set operator (AND/OR/UNLESS)
     pub(super) fn set_op_on_non_field_columns(
         &mut self,

src/query/src/promql/planner/test.rs

@@ -3188,6 +3188,46 @@ async fn ignoring_absent_label_keeps_tsid_binary_join() {
     assert!(!plan_str.contains("tag_1 ="), "{plan_str}");
 }
 
+#[tokio::test]
+async fn or_on_operand_with_matching_modifier_keeps_resolvable_time_index() {
+    // An operand of `or` derives its result labels from a matching modifier (`on(...)`,
+    // `ignoring(...)`, `group_left`/`group_right`), so the projected plan mixes qualifiers: the
+    // labels are `lhs.<label>` while the time index keeps the right operand's qualifier
+    // (`rhs.<time index>`). `or` must resolve that time index against the operand schema instead
+    // of reusing the qualifier of its first field.
+    for expr in [
+        "sum by (tag_0) (some_metric) or on (tag_0) \
+         sum by (tag_0) (some_metric) / ignoring(tag_1) group_left() sum by (tag_0) (some_metric)",
+        "sum by (tag_0) (some_metric) or on (tag_0) \
+         sum by (tag_0) (some_metric) / ignoring(tag_2) sum by (tag_0) (some_metric)",
+    ] {
+        let eval_stmt = build_eval_stmt(expr);
+        let table_provider = build_test_table_provider_with_tsid(
+            &[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())],
+            3,
+            1,
+        )
+        .await;
+        let plan =
+            PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
+                .await
+                .unwrap_or_else(|err| panic!("{expr}: {err}"));
+
+        let schema = plan.schema();
+        // The test table's time index column; the plan must carry it exactly once.
+        let time_index_columns = schema
+            .fields()
+            .iter()
+            .filter(|field| field.name() == "timestamp")
+            .count();
+        assert_eq!(1, time_index_columns, "{expr}: {schema:?}");
+        assert!(
+            schema.fields().iter().any(|field| field.name() == "tag_0"),
+            "{expr}: {schema:?}"
+        );
+    }
+}
+
 #[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");