fix: port five fixes to v1.1.0 (#15374, #15417, #15459, #15460, #15500) (#15526)

src/config/src/utils/async_file.rs

@@ -90,6 +90,8 @@ pub async fn put_file_contents(
     // Write to temporary file first
     let mut file_handle = File::create(&temp_file).await?;
     file_handle.write_all(contents).await?;
+    // tokio writes on a blocking task; without flush, rename can publish a short file
+    file_handle.flush().await?;
 
     // Atomically rename the temp file to the target file
     // This ensures we either have the old file or the new file, never a partially written file
@@ -368,6 +370,21 @@ mod tests {
 
     use super::*;
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_put_file_contents_is_complete_when_it_returns() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("large.bin");
+        // Past tokio's 2 MiB write buffer, so an unflushed tail is still in flight at the rename.
+        let content = vec![7u8; 4 << 20];
+        for _ in 0..20 {
+            put_file_contents(&path, &content).await.unwrap();
+            assert_eq!(
+                std::fs::metadata(&path).unwrap().len(),
+                content.len() as u64
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_get_file_contents_with_range() {
         let content = b"Hello World";

src/core/src/traces/red_insights.rs

@@ -29,7 +29,7 @@ pub const MAX_CREATES: usize = 15;
 
 const HISTOGRAM_INTERVAL: &str = "5m";
 const SCHEDULE_INTERVAL: &str = "1h";
-const DETECTION_WINDOW_SECONDS: i64 = 3_900;
+const DETECTION_WINDOW_SECONDS: i64 = 10_800;
 const MIN_BUCKET_REQUESTS: u32 = 20;
 const NAME_SEPARATOR: &str = " · ";
 /// Alert names forbid `/`; U+2215 reads the same and the web strip maps it back.

src/infra/src/cache/file_data/disk.rs

@@ -695,7 +695,12 @@ pub async fn get_opts(file: &str, options: GetOptions) -> object_store::Result<G
     options.check_preconditions(&meta)?;
 
     let range = match options.range {
-        Some(r) => r.as_range(meta.size).unwrap(),
+        Some(r) => r
+            .as_range(meta.size)
+            .map_err(|e| object_store::Error::Precondition {
+                path: file.to_string(),
+                source: Box::new(e),
+            })?,
         None => 0..meta.size,
     };
 
@@ -1537,6 +1542,28 @@ mod tests {
         ));
     }
 
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn get_opts_rejects_a_range_past_a_short_cached_file() {
+        let key = format!(
+            "files/default/logs/short/2026/10/15/10/{}.parquet",
+            config::ider::uuid()
+        );
+        set(&key, Bytes::from_static(b"0123456789abcdef"))
+            .await
+            .unwrap();
+        std::fs::write(get_file_path(&key).unwrap(), b"").unwrap();
+        let options = GetOptions {
+            range: Some((8..16).into()),
+            ..Default::default()
+        };
+        assert!(matches!(
+            get_opts(&key, options).await,
+            Err(object_store::Error::Precondition { .. })
+        ));
+        remove(&key).await.unwrap();
+    }
+
     #[test]
     fn key_escapes_rejects_traversal_and_absolute() {
         assert!(key_escapes("../etc/passwd"));

src/infra/src/cluster/mod.rs

@@ -426,15 +426,13 @@ pub async fn check_nodes_status(client: &reqwest::Client) -> Result<()> {
                 node.name,
                 node.http_addr
             );
-            let mut w = NODES_HEALTH_CHECK.write().await;
-            let Some(entry) = w.get_mut(&node.uuid) else {
-                // node haven't been added to the cluster yet, when the health check first succeed,
-                // it will be added to the check map
-                continue;
+            // count never-healthy peers too, or one that died during our startup is never removed
+            let times = {
+                let mut w = NODES_HEALTH_CHECK.write().await;
+                let entry = w.entry(node.uuid.clone()).or_insert(0);
+                *entry += 1;
+                *entry
             };
-            *entry += 1;
-            let times = *entry;
-            drop(w);
 
             if times >= cfg.health_check.failed_times {
                 log::error!(
@@ -808,4 +806,30 @@ mod tests {
         assert_eq!(nodes.len(), 3);
         assert_eq!(nodes, vec!["node-q-2", "node-q-5", "node-q-8"]);
     }
+
+    #[tokio::test]
+    async fn test_check_nodes_status_removes_never_healthy_node() {
+        let uuid = "test-never-healthy-node".to_string();
+        let dead = Node {
+            uuid: uuid.clone(),
+            name: "dead-ingester".to_string(),
+            http_addr: "http://127.0.0.1:1".to_string(),
+            role: vec![Role::Ingester],
+            status: NodeStatus::Online,
+            ..LOCAL_NODE.clone()
+        };
+        add_node_to_cache(dead).await;
+        assert!(!NODES_HEALTH_CHECK.read().await.contains_key(&uuid));
+
+        let client = reqwest::Client::new();
+        let failed_times = get_config().health_check.failed_times;
+        for _ in 1..failed_times {
+            check_nodes_status(&client).await.unwrap();
+        }
+        assert!(get_node_by_uuid(&uuid).await.is_some());
+
+        check_nodes_status(&client).await.unwrap();
+        assert!(get_node_by_uuid(&uuid).await.is_none());
+        assert!(!NODES_HEALTH_CHECK.read().await.contains_key(&uuid));
+    }
 }

src/promql_service/src/search/cache/mod.rs

@@ -223,6 +223,9 @@ pub async fn get(
         }
     }
 
+    // a series whose readings all fall before the window would reach callers with no samples
+    resp.series.retain(|series| !series.samples.is_empty());
+
     // if new_start == start, it means we have no data in cache, so we need to return None
     if new_start == start {
         return Ok(None);

src/promql_service/src/search/mod.rs

@@ -620,6 +620,8 @@ async fn merge_matrix_query(series: &[cluster_rpc::Series], org_id: &str) -> Res
     }
     let mut merged_data = merged_data
         .into_iter()
+        // Prometheus never answers a range query with a series that has no samples
+        .filter(|(_, samples)| !samples.is_empty())
         .map(|(sig, samples)| {
             let mut samples = samples
                 .into_iter()
@@ -873,6 +875,37 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_merge_matrix_drops_series_without_samples() {
+        let mut response = cluster_rpc::MetricsQueryResponse::default();
+        grpc::add_value(
+            &mut response,
+            Value::Range(RangeValue::new(
+                vec![Arc::new(Label::new("pod", "recovered"))],
+                Vec::<Sample>::new(),
+            )),
+        );
+        grpc::add_value(
+            &mut response,
+            Value::Range(RangeValue::new(
+                vec![Arc::new(Label::new("pod", "waiting"))],
+                vec![Sample::new(1_000_000, 1.0)],
+            )),
+        );
+
+        let value = merge_matrix_query(&response.series, "test_merge_empty_series")
+            .await
+            .unwrap();
+        let Value::Matrix(matrix) = value else {
+            panic!("expected matrix result");
+        };
+        assert_eq!(matrix.len(), 1);
+        assert_eq!(
+            matrix[0].labels,
+            vec![Arc::new(Label::new("pod", "waiting"))]
+        );
+    }
+
     #[tokio::test]
     async fn test_merge_exemplars_skips_series_without_exemplars() {
         let labels = vec![Arc::new(Label::new("job", "test"))];

src/search/src/datafusion/distributed_plan/remote_scan_exec.rs

@@ -26,7 +26,7 @@ use config::{
 use datafusion::{
     common::{DataFusionError, Result, tree_node::TreeNodeRecursion},
     execution::{SendableRecordBatchStream, TaskContext},
-    physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr},
+    physical_expr::{AcrossPartitions, EquivalenceProperties, Partitioning, PhysicalExpr},
     physical_plan::{
         DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
         execution_plan::{Boundedness, EmissionType},
@@ -77,7 +77,7 @@ impl RemoteScanExec {
         // Sharing happens on the leader only, a follower must never see a marker.
         let input = strip_shared_subplan_markers(input)?;
         let output_partitions = remote_scan_node.nodes.len();
-        let cache = Self::compute_properties(Arc::clone(&input.schema()), output_partitions);
+        let cache = Self::compute_properties(&input, output_partitions)?;
 
         // serialize the input plan and set it as the plan for the remote scan node
         let proto = get_physical_extension_codec();
@@ -162,17 +162,29 @@ impl RemoteScanExec {
 
     /// This function creates the cache object that stores the plan properties such as schema,
     /// equivalence properties, ordering, partitioning, etc.
-    fn compute_properties(schema: SchemaRef, n_partitions: usize) -> Arc<PlanProperties> {
-        let eq_properties = EquivalenceProperties::new(schema);
+    fn compute_properties(
+        input: &Arc<dyn ExecutionPlan>,
+        n_partitions: usize,
+    ) -> Result<Arc<PlanProperties>> {
+        let mut eq_properties = EquivalenceProperties::new(input.schema());
+        // Followers each run their own copy of the input; only literal constants hold on all.
+        eq_properties.add_constants(
+            input
+                .properties()
+                .eq_properties
+                .constants()
+                .into_iter()
+                .filter(|c| matches!(c.across_partitions, AcrossPartitions::Uniform(Some(_)))),
+        )?;
         let output_partitioning = Self::output_partitioning_helper(n_partitions);
-        Arc::new(PlanProperties::new(
+        Ok(Arc::new(PlanProperties::new(
             eq_properties,
             // Output Partitioning
             output_partitioning,
             // Execution Mode
             EmissionType::Incremental,
             Boundedness::Bounded,
-        ))
+        )))
     }
 
     pub fn set_analyze(mut self) -> Self {

src/search/src/datafusion/optimizer/physical_optimizer/remote_scan.rs

@@ -973,4 +973,109 @@ mod tests {
             assert_ne!(plan.name(), "RemoteScanExec");
         });
     }
+
+    async fn cluster_plan(sql: &str) -> (SessionContext, Arc<dyn ExecutionPlan>) {
+        use tracing_opentelemetry::OpenTelemetrySpanExt;
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("_timestamp", DataType::Int64, false),
+            Field::new("k8s_cluster", DataType::Utf8, true),
+            Field::new("k8s_resource_name", DataType::Utf8, true),
+            Field::new("k8s_namespace_name", DataType::Utf8, true),
+            Field::new("event_name", DataType::Utf8, true),
+            Field::new("value", DataType::Int64, true),
+        ]));
+        let mut file_id_lists = HashMap::new();
+        file_id_lists.insert(TableReference::from("t"), vec![]);
+        let nodes = (0..3)
+            .map(|_| Arc::new(config::meta::cluster::Node::default()) as Arc<dyn NodeInfo>)
+            .collect();
+        let remote_scan_nodes = Arc::new(RemoteScanNodes::new(
+            Request::default(),
+            nodes,
+            file_id_lists,
+            HashMap::new(),
+            false,
+            tracing::Span::current().context(),
+            None,
+        ));
+        let rule = RemoteScanRule {
+            remote_scan_nodes,
+            single_node_optimizer_enable: false,
+        };
+        let state = SessionStateBuilder::new()
+            .with_config(SessionConfig::new().with_target_partitions(4))
+            .with_runtime_env(Arc::new(RuntimeEnvBuilder::new().build().unwrap()))
+            .with_physical_optimizer_rule(Arc::new(rule))
+            .with_default_features()
+            .build();
+        let ctx = SessionContext::new_with_state(state);
+        let table = NewEmptyTable::new("t", schema)
+            .with_partitions(4)
+            .with_sort_order(crate::datafusion::sort_order::FileSortOrder::TimestampDesc);
+        ctx.register_table("t", Arc::new(table)).unwrap();
+        let logical = ctx.state().create_logical_plan(sql).await.unwrap();
+        let plan = ctx.state().create_physical_plan(&logical).await.unwrap();
+        (ctx, plan)
+    }
+
+    #[tokio::test]
+    async fn test_cluster_window_partitioned_by_filtered_constants_executes() {
+        use datafusion::physical_plan::windows::BoundedWindowAggExec;
+        let latest = "SELECT event_name, _timestamp FROM (SELECT event_name, _timestamp, \
+            ROW_NUMBER() OVER (PARTITION BY k8s_cluster, k8s_resource_name, \
+            COALESCE(k8s_namespace_name, ''), event_name ORDER BY _timestamp DESC) AS rn \
+            FROM t WHERE k8s_resource_name = 'nodes' AND k8s_cluster = 'c1'{terms}) \
+            WHERE rn = 1 ORDER BY _timestamp DESC LIMIT {limit}";
+        for (terms, limit) in [
+            ("", "5000"),
+            (" AND event_name = 'n1' AND k8s_namespace_name = 'ns1'", "1"),
+        ] {
+            let sql = latest.replace("{terms}", terms).replace("{limit}", limit);
+            let (ctx, plan) = cluster_plan(&sql).await;
+            let mut windows = vec![];
+            plan.apply(|node| {
+                if node.downcast_ref::<BoundedWindowAggExec>().is_some() {
+                    windows.push(Arc::clone(node));
+                }
+                Ok(TreeNodeRecursion::Continue)
+            })
+            .unwrap();
+            assert_eq!(windows.len(), 1, "{sql}");
+            assert!(decode_every_remote_scan(&ctx, &plan) >= 1, "{sql}");
+            if let Err(e) = windows[0].execute(0, ctx.task_ctx()) {
+                panic!("{sql}: {e}");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_cluster_union_literal_column_keeps_sort_requirement() {
+        let sql = "SELECT event_name, value, 'a' AS origin FROM t WHERE value > 1 \
+            UNION ALL SELECT event_name, value, 'b' AS origin FROM t WHERE value < 0 \
+            ORDER BY origin, value DESC LIMIT 15";
+        let (ctx, plan) = cluster_plan(sql).await;
+        assert!(decode_every_remote_scan(&ctx, &plan) >= 2);
+        plan.apply(|node| {
+            for (child, required) in node
+                .children()
+                .into_iter()
+                .zip(node.required_input_ordering())
+            {
+                // A merge over a remote scan trusts each follower's stream to be sorted.
+                let Some(required) =
+                    required.filter(|_| child.downcast_ref::<RemoteScanExec>().is_none())
+                else {
+                    continue;
+                };
+                let (alternatives, _) = required.into_alternatives();
+                let eq = child.equivalence_properties();
+                let satisfied = alternatives
+                    .iter()
+                    .any(|r| eq.ordering_satisfy_requirement(r.iter().cloned()).unwrap());
+                assert!(satisfied, "{} <- {}", node.name(), child.name());
+            }
+            Ok(TreeNodeRecursion::Continue)
+        })
+        .unwrap();
+    }
 }