Merge pull request #7744 from drmingdrmer/11-refact-read-table

BohuTANG · web-flow · commit a74637c08247 · 2022-09-20T22:17:01.000+08:00
chore: minor refactor for extract_scan_fields_from_projection(); cleanup info log
diff --git a/src/meta/raft-store/src/state_machine/sm.rs b/src/meta/raft-store/src/state_machine/sm.rs
@@ -263,11 +263,7 @@ impl StateMachine {
     /// command safely in case of network failure etc.
     #[tracing::instrument(level = "debug", skip(self, entry), fields(log_id=%entry.log_id))]
     pub async fn apply(&self, entry: &Entry<LogEntry>) -> Result<AppliedState, MetaStorageError> {
-        info!(
-            "apply: summary: {}; payload: {:?}",
-            entry.summary(),
-            entry.payload
-        );
+        info!("apply: summary: {}", entry.summary(),);
 
         let log_id = &entry.log_id;
 
@@ -280,8 +276,11 @@ impl StateMachine {
             txn_sm_meta.insert(&LastApplied, &StateMachineMetaValue::LogId(*log_id))?;
 
             match entry.payload {
-                EntryPayload::Blank => {}
+                EntryPayload::Blank => {
+                    info!("apply: blank");
+                }
                 EntryPayload::Normal(ref data) => {
+                    info!("apply: {}", data);
                     if let Some(ref txid) = data.txid {
                         let (serial, resp) =
                             self.txn_get_client_last_resp(&txid.client, &txn_tree)?;
@@ -303,7 +302,12 @@ impl StateMachine {
                     };
 
                     let res = self.apply_cmd(&data.cmd, &txn_tree, kv_pairs.as_ref(), log_time_ms);
-                    info!("apply_result: summary: {}; res: {:?}", entry.summary(), res);
+                    if let Ok(ok) = &res {
+                        info!("apply_result: summary: {}; res ok: {}", entry.summary(), ok);
+                    }
+                    if let Err(err) = &res {
+                        info!("apply_result: summary: {}; res err: {:?}", entry.summary(), err);
+                    }
 
                     let applied_state = res?;
 
@@ -317,6 +321,7 @@ impl StateMachine {
                     return Ok(Some(applied_state));
                 }
                 EntryPayload::Membership(ref mem) => {
+                    info!("apply: membership: {:?}", mem);
                     txn_sm_meta.insert(
                         &LastMembership,
                         &StateMachineMetaValue::Membership(EffectiveMembership {
@@ -760,7 +765,7 @@ impl StateMachine {
         kv_pairs: Option<&(DeleteByPrefixKeyMap, DeleteByPrefixKeyMap)>,
         log_time_ms: u64,
     ) -> Result<AppliedState, MetaStorageError> {
-        info!("apply_cmd: {:?}", cmd);
+        info!("apply_cmd: {}", cmd);
 
         match cmd {
             Cmd::IncrSeq { ref key } => self.apply_incr_seq_cmd(key, txn_tree),
diff --git a/src/meta/service/src/api/grpc/grpc_service.rs b/src/meta/service/src/api/grpc/grpc_service.rs
@@ -255,7 +255,7 @@ impl MetaService for MetaServiceImpl {
 
         let request = request.into_inner();
 
-        info!("Receive txn_request: {:?}", request);
+        info!("Receive txn_request: {}", request);
 
         let body = self.execute_txn(request).await;
         incr_meta_metrics_meta_sent_bytes(body.encoded_len() as u64);
diff --git a/src/meta/service/src/meta_service/meta_leader.rs b/src/meta/service/src/meta_service/meta_leader.rs
@@ -184,11 +184,19 @@ impl<'a> MetaLeader<'a> {
         // report metrics
         let _guard = WithCount::new((), ProposalPending);
 
-        info!("write LogEntry: {:?}", entry);
-        let write_rst = self.meta_node.raft.client_write(entry).await;
-        info!("raft.client_write rst: {:?}", write_rst);
+        info!("write LogEntry: {}", entry);
+        let write_res = self.meta_node.raft.client_write(entry).await;
+        if let Ok(ok) = &write_res {
+            info!(
+                "raft.client_write res ok: log_id: {}, data: {}, membership: {:?}",
+                ok.log_id, ok.data, ok.membership
+            );
+        }
+        if let Err(err) = &write_res {
+            info!("raft.client_write res err: {:?}", err);
+        }
 
-        match write_rst {
+        match write_res {
             Ok(resp) => Ok(resp.data),
             Err(cli_write_err) => Err(RaftWriteError::from_raft_err(cli_write_err)),
         }
diff --git a/src/meta/types/src/applied_state.rs b/src/meta/types/src/applied_state.rs
@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::fmt::Debug;
+use std::fmt::Formatter;
 
 use openraft::AppDataResponse;
 use serde::Deserialize;
@@ -47,6 +49,29 @@ pub enum AppliedState {
 
 impl AppDataResponse for AppliedState {}
 
+impl fmt::Display for AppliedState {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "AppliedState: ")?;
+        match self {
+            AppliedState::Seq { seq } => {
+                write!(f, "Seq: {}", seq)
+            }
+            AppliedState::Node { prev, result } => {
+                write!(f, "Node: prev: {:?}, result: {:?}", prev, result)
+            }
+            AppliedState::KV(change) => {
+                write!(f, "KV: {}", change)
+            }
+            AppliedState::TxnReply(txnreply) => {
+                write!(f, "Txn: {}", txnreply)
+            }
+            AppliedState::None => {
+                write!(f, "None")
+            }
+        }
+    }
+}
+
 impl AppliedState {
     /// Whether the state changed
     pub fn changed(&self) -> bool {
diff --git a/src/meta/types/src/change.rs b/src/meta/types/src/change.rs
@@ -12,6 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Debug;
+use std::fmt::Display;
+use std::fmt::Formatter;
+
 use serde::Deserialize;
 use serde::Serialize;
 
@@ -104,3 +108,15 @@ where
         unreachable!("impossible: both prev and result are None");
     }
 }
+
+impl<T, ID> Display for Change<T, ID>
+where
+    T: Debug + Clone + PartialEq,
+    ID: Debug + Clone + PartialEq,
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "id: {:?}", self.ident)?;
+        write!(f, "prev: {:?}", self.prev)?;
+        write!(f, "result: {:?}", self.result)
+    }
+}
diff --git a/src/meta/types/src/cmd.rs b/src/meta/types/src/cmd.rs
@@ -79,7 +79,7 @@ impl fmt::Display for Cmd {
                 write!(f, "upsert_kv:{}", upsert_kv)
             }
             Cmd::Transaction(txn) => {
-                write!(f, "txn:{:?}", txn)
+                write!(f, "txn:{}", txn)
             }
         }
     }
diff --git a/src/meta/types/src/log_entry.rs b/src/meta/types/src/log_entry.rs
@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Display;
+use std::fmt::Formatter;
+
 use openraft::AppData;
 use serde::Deserialize;
 use serde::Serialize;
@@ -42,3 +45,16 @@ pub struct LogEntry {
 }
 
 impl AppData for LogEntry {}
+
+impl Display for LogEntry {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        if let Some(txid) = &self.txid {
+            write!(f, "txid: {:?}", txid)?;
+        }
+        if let Some(time) = &self.time_ms {
+            write!(f, "time: {} ms", time)?;
+        }
+
+        write!(f, " cmd: {}", self.cmd)
+    }
+}
diff --git a/src/query/service/src/sql/executor/physical_plan_builder.rs b/src/query/service/src/sql/executor/physical_plan_builder.rs
@@ -126,7 +126,7 @@ impl PhysicalPlanBuilder {
                     }
                     if let Some(prewhere) = &scan.prewhere {
                         // if there is a prewhere optimization,
-                        // we can prune `PhysicalScan`'s ouput schema.
+                        // we can prune `PhysicalScan`'s output schema.
                         if prewhere.output_columns.contains(index) {
                             name_mapping.insert(column.name().to_string(), index.to_string());
                         }
diff --git a/src/query/service/src/storages/storage_table_read_plan.rs b/src/query/service/src/storages/storage_table_read_plan.rs
@@ -16,12 +16,12 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 
 use common_datavalues::DataField;
+use common_datavalues::DataSchema;
 use common_exception::Result;
 use common_legacy_planners::Extras;
 use common_legacy_planners::Projection;
 use common_legacy_planners::ReadDataSourcePlan;
 use common_legacy_planners::SourceInfo;
-use common_meta_app::schema::TableInfo;
 
 use crate::sessions::QueryContext;
 use crate::storages::Table;
@@ -55,16 +55,21 @@ impl ToReadDataSourcePlan for dyn Table {
         push_downs: Option<Extras>,
     ) -> Result<ReadDataSourcePlan> {
         let (statistics, parts) = self.read_partitions(ctx, push_downs.clone()).await?;
+
         let table_info = self.get_table_info();
+        let table_meta = &table_info.meta;
         let description = statistics.get_description(table_info);
 
         let scan_fields = match (self.benefit_column_prune(), &push_downs) {
             (true, Some(push_downs)) => match &push_downs.prewhere {
-                Some(prewhere) => {
-                    extract_scan_fields_from_projection(table_info, &prewhere.output_columns)
-                }
+                Some(prewhere) => extract_scan_fields_from_projection(
+                    &table_meta.schema,
+                    &prewhere.output_columns,
+                ),
                 _ => match &push_downs.projection {
-                    Some(projection) => extract_scan_fields_from_projection(table_info, projection),
+                    Some(projection) => {
+                        extract_scan_fields_from_projection(&table_meta.schema, projection)
+                    }
                     _ => None,
                 },
             },
@@ -87,15 +92,13 @@ impl ToReadDataSourcePlan for dyn Table {
 }
 
 fn extract_scan_fields_from_projection(
-    table_info: &TableInfo,
+    schema: &DataSchema,
     projection: &Projection,
 ) -> Option<BTreeMap<usize, DataField>> {
     match projection {
         Projection::Columns(ref indices) => {
-            if indices.len() < table_info.schema().fields().len() {
-                let fields = indices
-                    .iter()
-                    .map(|i| table_info.schema().field(*i).clone());
+            if indices.len() < schema.fields().len() {
+                let fields = indices.iter().map(|i| schema.field(*i).clone());
 
                 Some((indices.iter().cloned().zip(fields)).collect::<BTreeMap<_, _>>())
             } else {
@@ -104,7 +107,7 @@ fn extract_scan_fields_from_projection(
         }
         Projection::InnerColumns(ref path_indices) => {
             let column_ids: Vec<usize> = path_indices.keys().cloned().collect();
-            let new_schema = table_info.schema().inner_project(path_indices);
+            let new_schema = schema.inner_project(path_indices);
             Some(
                 (column_ids.iter().cloned().zip(new_schema.fields().clone()))
                     .collect::<BTreeMap<_, _>>(),

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ impl fmt::Display for Cmd {`
`79`	`79`	`write!(f, "upsert_kv:{}", upsert_kv)`
`80`	`80`	`}`
`81`	`81`	`Cmd::Transaction(txn) => {`
`82`		`- write!(f, "txn:{:?}", txn)`
	`82`	`+ write!(f, "txn:{}", txn)`
`83`	`83`	`}`
`84`	`84`	`}`
`85`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -126,7 +126,7 @@ impl PhysicalPlanBuilder {`
`126`	`126`	`}`
`127`	`127`	`if let Some(prewhere) = &scan.prewhere {`
`128`	`128`	`// if there is a prewhere optimization,`
`129`		- // we can prune `PhysicalScan`'s ouput schema.
	`129`	+ // we can prune `PhysicalScan`'s output schema.
`130`	`130`	`if prewhere.output_columns.contains(index) {`
`131`	`131`	`name_mapping.insert(column.name().to_string(), index.to_string());`
`132`	`132`	`}`