feat: able to set read preference to flownode (#6696)

fix: correctly compare the opened follower regions in startup

Signed-off-by: luofucong <luofc@foxmail.com>

src/common/session/Cargo.toml

@@ -8,4 +8,5 @@ license.workspace = true
 workspace = true
 
 [dependencies]
+serde.workspace = true
 strum.workspace = true

src/common/session/src/lib.rs

@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use serde::{Deserialize, Serialize};
 use strum::{AsRefStr, Display, EnumString};
 
 /// Defines the read preference for frontend route operations,
 /// determining whether to read from the region leader or follower.
-#[derive(Debug, Clone, Copy, Default, EnumString, Display, AsRefStr, PartialEq, Eq)]
+#[derive(
+    Debug, Clone, Copy, Default, EnumString, Display, AsRefStr, PartialEq, Serialize, Deserialize,
+)]
 pub enum ReadPreference {
     #[default]
     // Reads all operations from the region leader. This is the default mode.

src/datanode/src/datanode.rs

@@ -705,6 +705,7 @@ async fn open_all_regions(
             follower_regions.len()
         );
         let mut region_requests = Vec::with_capacity(follower_regions.len());
+        let num_regions = follower_regions.len();
         for (region_id, engine, store_path, options) in follower_regions {
             let table_dir = table_dir(&store_path, region_id.table_id());
             region_requests.push((

src/flow/src/batching_mode.rs

@@ -17,6 +17,7 @@
 use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
+use session::ReadPreference;
 
 pub(crate) mod engine;
 pub(crate) mod frontend_client;
@@ -54,6 +55,8 @@ pub struct BatchingModeOptions {
     pub experimental_max_filter_num_per_query: usize,
     /// Time window merge distance
     pub experimental_time_window_merge_threshold: usize,
+    /// Read preference of the Frontend client.
+    pub read_preference: ReadPreference,
 }
 
 impl Default for BatchingModeOptions {
@@ -68,6 +71,7 @@ impl Default for BatchingModeOptions {
             experimental_frontend_activity_timeout: Duration::from_secs(60),
             experimental_max_filter_num_per_query: 20,
             experimental_time_window_merge_threshold: 3,
+            read_preference: Default::default(),
         }
     }
 }

src/flow/src/batching_mode/frontend_client.rs

@@ -36,6 +36,7 @@ use rand::rng;
 use rand::seq::SliceRandom;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::{QueryContextBuilder, QueryContextRef};
+use session::hints::READ_PREFERENCE_HINT;
 use snafu::{OptionExt, ResultExt};
 
 use crate::batching_mode::BatchingModeOptions;
@@ -363,7 +364,10 @@ impl FrontendClient {
                     .handle_with_retry(
                         req.clone(),
                         batch_opts.experimental_grpc_max_retries,
-                        &[(QUERY_PARALLELISM_HINT, &query.parallelism.to_string())],
+                        &[
+                            (QUERY_PARALLELISM_HINT, &query.parallelism.to_string()),
+                            (READ_PREFERENCE_HINT, batch_opts.read_preference.as_ref()),
+                        ],
                     )
                     .await
                     .with_context(|_| InvalidRequestSnafu {