You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: hash_agg.rs
    path of file: ./repos/risingwave/src/stream/src/executor
    
    the code of the file until where you have to start completion: // Copyright 2024 RisingWave Labs
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    
    
    async fn flush_data<'a>(this: &'a mut ExecutorInner<K, S>, vars: &'a mut ExecutionVars<K, S>) {
        let window_watermark = vars.window_watermark.take();

        // flush changed states into intermediate state table
        for agg_group in vars.dirty_groups.values() {
            let encoded_states = agg_group.encode_states(&this.agg_funcs)?;
            if this.emit_on_window_close {
                vars.buffer
                    .update_without_old_value(encoded_states, &mut this.intermediate_state_table);
            } else {
                this.intermediate_state_table
                    .update_without_old_value(encoded_states);
            }
        }

        if this.emit_on_window_close {
            // remove all groups under watermark and emit their results
            if let Some(watermark) = window_watermark.as_ref() {
                #[for_await]
                for row in vars
                    .buffer
                    .consume(watermark.clone(), &mut this.intermediate_state_table)
                {
                    let row = row?;
                    let group_key = row
                        .clone()
                        .into_iter()
                        .take(this.group_key_indices.len())
                        .collect();
                    let states = row.into_iter().skip(this.group_key_indices.len()).collect();

                    let mut agg_group = AggGroup::create_eowc(
                        this.version,
                        Some(GroupKey::new(
                            group_key,
                            Some(this.group_key_table_pk_projection.clone()),
                        )),
                        &this.agg_calls,
                        &this.agg_funcs,
                        &this.storages,
                        &states,
                        &this.input_pk_indices,
                        this.row_count_index,
                        this.extreme_cache_size,
                        &this.input_schema,
                    )?;

                    let change = agg_group
                        .build_change(&this.storages, &this.agg_funcs)
                        .await?;
                    if let Some(change) = change {
                        if let Some(chunk) = vars.chunk_builder.append_record(change) {
                            yield chunk;
                        }
                    }
                }
            }
        } else {
            // emit on update
            // TODO(wrj,rc): we may need to parallelize it and set a reasonable concurrency limit.
            for mut agg_group in vars.dirty_groups.values_mut() {
                let agg_group = agg_group.as_mut();
                let change = agg_group
                    .build_change(&this.storages, &this.agg_funcs)
                    .await?;
                if let Some(change) = change {
                    if let Some(chunk) = vars.chunk_builder.append_record(change) {
                        yield chunk;
                    }
                }
            }
        }

        // move dirty groups back to cache
        for (key, agg_group) in vars.dirty_groups.drain() {
            vars.agg_group_cache.put(key, Some(agg_group));
        }

        // Yield the remaining rows in chunk builder.
        if let Some(chunk) = vars.chunk_builder.take() {
            yield chunk;
        }

        if let Some(watermark) = window_watermark {
            // Update watermark of state tables, for state cleaning.
            this.all_state_tables_mut()
                .for_each(|table| table.update_watermark(watermark.clone(), false));
        }

        // Flush distinct dedup state.
        vars.distinct_dedup.flush(&mut this.distinct_dedup_tables)?;

        // Evict cache to target capacity.
        vars.agg_group_cache.evict();
    }