#[cfg(test)]
mod tests {
    use std::fmt::{Display, Formatter};
    use arrow::array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray};
    use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
    use datafusion::common::Result;
    use datafusion::execution::memory_pool::FairSpillPool;
    use datafusion::execution::runtime_env::RuntimeEnvBuilder;
    use datafusion::execution::TaskContext;
    use datafusion::functions_aggregate::sum::sum_udaf;
    use datafusion::physical_expr::aggregate::AggregateExprBuilder;
    use datafusion::physical_expr::expressions::{lit, Column};
    use datafusion::physical_plan::aggregates::{PhysicalGroupBy, AggregateExec, AggregateMode};
    use datafusion::physical_plan::common::collect;
    use datafusion::physical_plan::ExecutionPlan;
    use rand::{random, thread_rng, Rng};
    use std::sync::Arc;
    use datafusion_physical_plan::memory::{LazyBatchGenerator, LazyMemoryExec};
    use parking_lot::RwLock;

    #[tokio::test]
    async fn test_debug() -> Result<()> {
        let scan_schema = Arc::new(Schema::new(vec![
            Field::new("col_0", DataType::Int64, true),
            Field::new("col_1", DataType::Utf8, true),
            Field::new("col_2", DataType::Utf8, true),
            Field::new("col_3", DataType::Utf8, true),
            Field::new("col_4", DataType::Utf8, true),
            Field::new("col_5", DataType::Int32, true),
            Field::new("col_6", DataType::Utf8, true),
            Field::new("col_7", DataType::Utf8, true),
            Field::new("col_8", DataType::Utf8, true),
        ]));

        let group_by = PhysicalGroupBy::new_single(vec![
            (Arc::new(Column::new("col_1", 1)), "col_1".to_string()),
            (Arc::new(Column::new("col_7", 7)), "col_7".to_string()),
            (Arc::new(Column::new("col_0", 0)), "col_0".to_string()),
            (Arc::new(Column::new("col_8", 8)), "col_8".to_string()),
        ]);

        fn generate_int64_array() -> ArrayRef {
            Arc::new(Int64Array::from_iter_values(
                (0..8192).map(|_| random::<i64>()),
            ))
        }
        fn generate_int32_array() -> ArrayRef {
            Arc::new(Int32Array::from_iter_values(
                (0..8192).map(|_| random::<i32>()),
            ))
        }

        fn generate_string_array() -> ArrayRef {
            Arc::new(StringArray::from(
                (0..8192)
                    .map(|_| -> String {
                        thread_rng()
                            .sample_iter::<char, _>(rand::distributions::Standard)
                            .take(10)
                            .collect()
                    })
                    .collect::<Vec<_>>(),
            ))
        }

        fn generate_record_batch(schema: &SchemaRef) -> Result<RecordBatch> {
            RecordBatch::try_new(
                Arc::clone(&schema),
                vec![
                    generate_int64_array(),
                    generate_string_array(),
                    generate_string_array(),
                    generate_string_array(),
                    generate_string_array(),
                    generate_int32_array(),
                    generate_string_array(),
                    generate_string_array(),
                    generate_string_array(),
                ],
            )
            .map_err(|err| err.into())
        }

        let aggregate_expressions = vec![Arc::new(
            AggregateExprBuilder::new(sum_udaf(), vec![lit(1i64)])
                .schema(Arc::clone(&scan_schema))
                .alias("SUM(1i64)")
                .build()?,
        )];

        #[derive(Debug)]
        struct Generator {
            index: usize,
            count: usize,
            schema: SchemaRef,
        }

        impl Display for Generator {
            fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
                write!(f, "Generator")
            }
        }

        impl LazyBatchGenerator for Generator {
            fn generate_next_batch(&mut self) -> Result<Option<RecordBatch>> {
                if self.index > self.count {
                    return Ok(None);
                }

                let batch = generate_record_batch(&self.schema)?;
                self.index += 1;

                Ok(Some(batch))
            }
        }

        let generator = Generator {
            index: 0,
            count: 10,
            schema: Arc::clone(&scan_schema),
        };

        let plan: Arc<dyn ExecutionPlan> =
          Arc::new(LazyMemoryExec::try_new(Arc::clone(&scan_schema), vec![Arc::new(RwLock::new(generator))])?);

        let single_aggregate = Arc::new(AggregateExec::try_new(
            AggregateMode::Single,
            group_by,
            aggregate_expressions.clone(),
            vec![None; aggregate_expressions.len()],
            plan,
            Arc::clone(&scan_schema),
        )?);

        let memory_pool = Arc::new(FairSpillPool::new(10006216));
        let task_ctx = Arc::new(
            TaskContext::default().with_runtime(Arc::new(
                RuntimeEnvBuilder::new()
                    .with_memory_pool(memory_pool)
                    .build()?,
            )),
        );

        let res = collect(single_aggregate.execute(0, Arc::clone(&task_ctx))?).await;

        match res {
            Ok(_) => println!("Success"),
            Err(e) => {
                println!("Error: {}", e);

                return Err(e);
            },
        }

        Ok(())
    }
}