add retention example work with parquet file

examples/retention_parquet.rs

@@ -0,0 +1,51 @@
+use std::sync::Arc;
+
+use datafusion::arrow::util::pretty::print_batches;
+use datafusion::datasource::MemTable;
+use datafusion::error::Result;
+use datafusion::prelude::*;
+use datafusion_uba::retention::{create_retention_count, create_retention_sum};
+use datafusion_uba::test_util;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let ctx = SessionContext::new();
+
+    ctx.register_parquet(
+        "event",
+        format!("{}/event.parquet", test_util::parquet_test_data()).as_str(),
+        Default::default(),
+    )
+    .await
+    .unwrap();
+
+    ctx.register_udaf(create_retention_count());
+    ctx.register_udaf(create_retention_sum());
+
+    let df = ctx
+        .sql(
+            "select distinct_id,retention_count(\
+                    case when xwhat='$startup' then true else false end,\
+                    case when xwhat='$pageview' then true else false end,\
+                    20230107-20230101,\
+                    ds-20230101 \
+                    ) as stats \
+                from event group by distinct_id order by distinct_id",
+        )
+        .await?;
+    let results = df.clone().collect().await?;
+    // print_batches(&results);
+
+    let provider = MemTable::try_new(df.schema().clone().into(), vec![results])?;
+    ctx.register_table("retention_count_result", Arc::new(provider))?;
+
+    let results = ctx
+        .sql("select retention_sum(stats) from retention_count_result")
+        .await?
+        .collect()
+        .await?;
+
+    print_batches(&results)?;
+
+    Ok(())
+}

src/retention/retention_sum.rs

@@ -131,7 +131,6 @@ impl Accumulator for RetentionSum {
     fn evaluate(&self) -> Result<ScalarValue> {
         let arr_ref = &ScalarValue::iter_to_array(self.total_active.clone()).unwrap();
         let mut final_result: Vec<Vec<ScalarValue>> = Vec::new();
-
         for index in 0..arr_ref.len() {
             if let ScalarValue::List(Some(per_user), _) =
                 ScalarValue::try_from_array(arr_ref, index)?