feat: flownode inc query checkpoint

[discord9]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

What's changed and what's your intention?

PR Checklist

Please convert it to a draft if some of the following conditions are not met.

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR requires documentation updates.
• API changes are backward compatible.
• Schema or data changes are backward compatible.

[gemini-code-assist]

Code Review

This pull request implements incremental read support for batching flows by introducing a state machine that manages transitions between full snapshot and incremental modes based on region watermarks. Key changes include logic to analyze and rewrite aggregate plans into delta-left-join-sink merges, extensions to the FrontendClient for handling terminal metrics, and the addition of comprehensive integration tests. Feedback from the review focuses on improving the robustness of error handling by replacing brittle string parsing with structured error variants and simplifying the logic for validating incremental checkpoint advancement to remove redundant checks.

[gemini-code-assist]

Parsing the debug representation of an error (format!("{err:?}")) is brittle and can easily break if the error formatting changes in the future. It would be more robust to introduce a specific error variant for "stale cursor" and propagate it as a structured error. This is especially important for correctly identifying connection-related errors, as only those should trigger a connection reset. While the TODO comment acknowledges this, this should be prioritized to improve the robustness of error handling.

References

1. Only reset a client connection for connection-related errors. Robust error identification is necessary to follow this rule.

[gemini-code-assist]

This function's logic is a bit complex and can be simplified for better readability and maintainability. The check participating_regions.iter().all(|region_id| self.checkpoints.contains_key(region_id)) is redundant, as the subsequent check already covers this. Using a match statement can also make the logic more explicit and easier to understand than zip().is_some_and().

    pub fn can_advance_incremental_checkpoints_with_participation(
        &self,
        participating_regions: &BTreeSet<u64>,
        watermark_map: &HashMap<u64, u64>,
    ) -> bool {
        !self.checkpoints.is_empty()
            && !participating_regions.is_empty()
            && participating_regions.len() == watermark_map.len()
            && participating_regions.iter().all(|region_id| {
                match (self.checkpoints.get(region_id), watermark_map.get(region_id)) {
                    (Some(checkpoint), Some(seq)) => seq >= checkpoint,
                    _ => false,
                }
            })
    }