fn f_up(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion> {
    let name = node.name();
    if name == "NewEmptyExec" {
        let table = node.as_any().downcast_ref::<NewEmptyExec>().unwrap();
        self.table_name = Some(TableReference::from(table.name()));
        Ok(TreeNodeRecursion::Continue)
    } else {
        Ok(TreeNodeRecursion::Continue)
    }

fn f_up(&mut self, plan: Self::Node) -> Result<Transformed<Self::Node>> {
    if plan
        .as_any()
        .downcast_ref::<SortPreservingMergeExec>()
        .is_some()
    {
        // Get the index fields of the underlying table
        let mut visitor = TableNameVisitor::new();
        plan.visit(&mut visitor)?;
        let Some(table_name) = visitor.table_name else {
            return Ok(Transformed::new(plan, false, TreeNodeRecursion::Stop));
        };
        let index_fields = self
            .index_fields
            .get(&table_name)
            .cloned()
            .unwrap_or(HashSet::new());

        // check if the query is simple topn or simple distinct
        if let Some(index_optimize_mode) =
            is_simple_topn(Arc::clone(&plan), index_fields.clone())
        {
            // Check for SimpleTopN
            let mut rewriter = IndexOptimizerRewrite::new(index_optimize_mode);
            let plan = plan.rewrite(&mut rewriter)?.data;
            return Ok(Transformed::new(plan, true, TreeNodeRecursion::Stop));
        } else if let Some(index_optimize_mode) =
            is_simple_distinct(Arc::clone(&plan), index_fields.clone())
        {
            // Check for SimpleDistinct
            let mut rewriter = IndexOptimizerRewrite::new(index_optimize_mode);
            let plan = plan.rewrite(&mut rewriter)?.data;
            return Ok(Transformed::new(plan, true, TreeNodeRecursion::Stop));
        } else {
            return Ok(Transformed::new(plan, false, TreeNodeRecursion::Continue));
        }
    }

// should after remote scan
let mut index_fields: HashMap<TableReference, HashSet<String>> = HashMap::new();
for (stream_name, schema) in sql.schemas.iter() {
    let stream_settings = infra::schema::unwrap_stream_settings(schema.schema());
    let idx_fields = get_stream_setting_index_fields(&stream_settings);
    let idx_fields = idx_fields
        .into_iter()
        .filter_map(|index_field| {
            if schema.contains_field(&index_field) {
                Some(index_field)
            } else {
                None
            }
        })
        .collect::<HashSet<_>>();
    index_fields.insert(stream_name.clone(), idx_fields);
}
rules.push(Arc::new(LeaderIndexOptimizerRule::new(index_fields)) as _);