fix(query): match the raw scan's own field and engine rules

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

src/query/src/promql/planner.rs

@@ -5507,16 +5507,19 @@ impl PromPlanner {
 
     /// Whether every field column carries a type the region's raw series preflight serves.
     ///
-    /// That preflight accepts numeric columns and native histograms only, so anything else - a
-    /// string, a boolean, a timestamp - has no raw framing. A route that offered it would fail
-    /// after the logical plan is fixed; declining keeps the existing plan.
+    /// The preflight decides on `ConcreteDataType::is_numeric`, which excludes the decimals, so
+    /// the Arrow type is converted rather than asked: Arrow's own `is_numeric` includes
+    /// `Decimal128`/`Decimal256` (and `Float16`, which has no column type here), and a decimal
+    /// field would then pass here and fail once the logical plan is fixed. Anything else - a
+    /// string, a boolean, a timestamp - has no raw framing either.
     fn field_columns_are_raw_series_servable(
         schema: &DFSchemaRef,
         field_columns: &[String],
     ) -> bool {
         field_columns.iter().all(|field| {
             Self::field_column_type(schema, field).is_some_and(|data_type| {
-                data_type.is_numeric() || data_type == &Self::native_histogram_arrow_type()
+                datatypes::prelude::ConcreteDataType::from_arrow_type(data_type).is_numeric()
+                    || data_type == &Self::native_histogram_arrow_type()
             })
         })
     }

src/query/src/promql/planner/native.rs

@@ -30,7 +30,7 @@
 use std::sync::Arc;
 use std::time::{Duration, UNIX_EPOCH};
 
-use common_catalog::consts::{MITO_ENGINE, MITO2_ENGINE};
+use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE, MITO2_ENGINE};
 use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, TimeUnit as ArrowTimeUnit};
 use datafusion::common::{DFSchema, DFSchemaRef};
 use datafusion::error::DataFusionError;
@@ -385,15 +385,23 @@ fn selector_range_millis(expr: &PromExpr) -> Option<i64> {
 /// Whether `engine` can serve raw series frames to the node this route builds.
 ///
 /// A whitelist, not a probe: the logical planner cannot ask a region what its scanner supports
-/// without reading the table, and a false negative only costs the query its native route. Mito's
-/// series scanner implements the raw path; every other engine in the workspace leaves
-/// `raw_series_schema` at its default `None`, so only those two are named. This is deliberately not
-/// [`supports_pk_dictionary_encoding`]: that answers a different question and includes the metric
-/// engine, whose logical tables a bare scan cannot reach anyway.
+/// without reading the table, and a false negative only costs the query its native route.
+///
+/// Mito's series scanner implements the raw path. The metric engine is named too because a logical
+/// region reads through its physical region and hands back that same scanner - its own raw-series
+/// test is `test_raw_series_scan_uses_logical_metric_route_and_bound_metadata` - and the ordinary
+/// metric rewrite puts a filter and an alias between the scan and this node, which the bare-scan
+/// check above already refuses. A logical table reaches that check with a bare scan only when the
+/// physical table cannot be resolved, and there the logical scan is served by the same scanner.
+///
+/// Every other engine in the workspace leaves `raw_series_schema` at its default `None`: a
+/// file-engine table, for one, is read by a `SinglePartitionScanner` that cannot frame a series.
+///
+/// This is deliberately not [`supports_pk_dictionary_encoding`]: that answers a different question.
 ///
 /// [`supports_pk_dictionary_encoding`]: table::table::scan
 fn supports_native_raw_series(engine: &str) -> bool {
-    matches!(engine, MITO_ENGINE | MITO2_ENGINE)
+    matches!(engine, MITO_ENGINE | MITO2_ENGINE | METRIC_ENGINE)
 }
 
 /// Whether any node of `tree` reads the whole input vector at a point rather than one series.

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

@@ -18,7 +18,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use catalog::RegisterTableRequest;
 use catalog::memory::{MemoryCatalogManager, new_memory_catalog_manager};
 use common_base::Plugins;
-use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, FILE_ENGINE, MITO_ENGINE};
+use common_catalog::consts::{
+    DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, FILE_ENGINE, METRIC_ENGINE, MITO_ENGINE,
+};
 use common_query::native_histogram::{
     CUSTOM_BUCKETS_SCHEMA, CounterResetHint, NativeHistogram, build_histogram_array,
 };
@@ -946,7 +948,9 @@ async fn build_test_table_provider_with_field_type(
         ConcreteDataType::timestamp_millisecond_datatype(),
         false,
         field_type,
-        DEFAULT_ENGINE,
+        // The native engine, so the field guard below is what refuses this table: on the blank
+        // engine the engine gate would answer first and this test would prove nothing.
+        NATIVE_ENGINE,
     )
     .await
 }
@@ -1877,6 +1881,37 @@ async fn native_fused_tag_partitioned_table_still_plans_fused_series() {
 }
 
 /// The flag on and a shape the compiler refuses: the existing planner plans it unchanged.
+/// A metric-engine logical table is served by its physical region's scanner, so it is admitted.
+///
+/// The ordinary metric rewrite puts a filter and an alias between the scan and this node, which the
+/// bare-scan check refuses - that is why such a query usually falls back. This test pins the one
+/// shape that reaches the node bare (the physical table could not be resolved), so that the engine
+/// whitelist does not refuse a table whose region really does serve frames.
+#[tokio::test]
+async fn native_fused_metric_engine_bare_scan_is_admitted() {
+    let native = build_native_fused_plan_with_provider(
+        "rate(some_metric[5m])",
+        true,
+        build_test_table_provider_with_options(
+            &[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())],
+            1,
+            1,
+            vec![],
+            ConcreteDataType::timestamp_millisecond_datatype(),
+            false,
+            ConcreteDataType::float64_datatype(),
+            METRIC_ENGINE,
+        )
+        .await,
+    )
+    .await;
+    assert_eq!(
+        find_fused_series(&native).map(|fused| fused.promql().to_string()),
+        Some("rate(some_metric[5m])".to_string()),
+        "a metric-engine table serves frames, so a bare scan over one keeps the native route"
+    );
+}
+
 /// A table whose engine has no raw-series scanner keeps the existing plan.
 ///
 /// A file-engine table answers an ordinary read with a `SinglePartitionScanner`, whose
@@ -1938,6 +1973,9 @@ async fn native_fused_non_numeric_field_falls_back() {
     for field_type in [
         ConcreteDataType::string_datatype(),
         ConcreteDataType::boolean_datatype(),
+        // Arrow's `is_numeric` includes the decimals while the region's preflight does not, so
+        // asking Arrow would let a decimal field through to a planning error.
+        ConcreteDataType::decimal128_datatype(10, 2),
     ] {
         let native = build_native_fused_plan_with_provider(
             "some_metric",