diff --git a/Cargo.lock b/Cargo.lock
index 0f70678a143..e657fb2b82b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -10782,6 +10782,7 @@ dependencies = [
  "sqllogictest",
  "thiserror 2.0.18",
  "tokio",
+ "tracing-subscriber",
  "vortex",
  "vortex-datafusion",
  "vortex-duckdb",
diff --git a/vortex-duckdb/build.rs b/vortex-duckdb/build.rs
index 7647011e366..188d2d43ec5 100644
--- a/vortex-duckdb/build.rs
+++ b/vortex-duckdb/build.rs
@@ -27,11 +27,12 @@ const DEFAULT_DUCKDB_VERSION: &str = "1.5.3";
 
 const BUILD_ARTIFACTS: [&str; 3] = ["libduckdb.dylib", "libduckdb.so", "libduckdb_static.a"];
 
-const SOURCE_FILES: [&str; 7] = [
+const SOURCE_FILES: [&str; 8] = [
     "cpp/vortex_duckdb.cpp",
     "cpp/copy_function.cpp",
     "cpp/expr.cpp",
     "cpp/scalar_fn_pushdown.cpp",
+    "cpp/aggregate_fn_pushdown.cpp",
     "cpp/table_filter.cpp",
     "cpp/table_function.cpp",
     "cpp/vector.cpp",
diff --git a/vortex-duckdb/cpp/aggregate_fn_pushdown.cpp b/vortex-duckdb/cpp/aggregate_fn_pushdown.cpp
new file mode 100644
index 00000000000..f3c26e20f02
--- /dev/null
+++ b/vortex-duckdb/cpp/aggregate_fn_pushdown.cpp
@@ -0,0 +1,136 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+#include "aggregate_fn_pushdown.hpp"
+#include "duckdb/planner/expression/bound_aggregate_expression.hpp"
+#include "duckdb/planner/operator/logical_aggregate.hpp"
+#include "scalar_fn_pushdown.hpp"
+#include "table_function.hpp"
+
+using enum LogicalOperatorType;
+
+LogicalOperatorPtr TryPushdownAggregateFunctions(ClientContext &context, LogicalOperatorPtr plan) {
+    Analyses analyses;
+    Projections projections;
+    FindGetsAndProjections(*plan, analyses, projections);
+    if (analyses.empty()) {
+        return plan;
+    }
+    return RewriteAggregates(context, std::move(plan), analyses, projections);
+}
+
+LogicalOperatorPtr RewriteAggregates(ClientContext &context,
+                                     LogicalOperatorPtr op,
+                                     Analyses &analyses,
+                                     const Projections &projections) {
+    for (auto &child : op->children) {
+        child = RewriteAggregates(context, std::move(child), analyses, projections);
+    }
+    if (op->type == LOGICAL_AGGREGATE_AND_GROUP_BY) {
+        return TryReplaceAggregate(context, std::move(op), analyses, projections);
+    }
+    return op;
+}
+
+static bool IsUngrouped(const LogicalAggregate &agg) {
+    return agg.groups.empty() && agg.grouping_sets.empty() && agg.grouping_functions.empty() &&
+           !agg.expressions.empty();
+}
+
+constexpr inline idx_t COUNT_STAR_PROJ_IDX = std::numeric_limits<TableColumnStorageIndex>::max();
+
+LogicalOperatorPtr TryReplaceAggregate(ClientContext &context,
+                                       LogicalOperatorPtr op,
+                                       Analyses &analyses,
+                                       const Projections &projections) {
+    LogicalAggregate &agg = op->Cast<LogicalAggregate>();
+    if (!IsUngrouped(agg)) {
+        return op;
+    }
+
+    LogicalGet *const get = GetChildGet(agg);
+    if (get == nullptr) {
+        return op;
+    }
+
+    vector<std::pair<TableColumnStorageIndex, const Expression &>> input;
+    const idx_t N = agg.expressions.size();
+    input.reserve(N);
+
+    for (const auto &expr : agg.expressions) {
+        if (expr->GetExpressionClass() != ExpressionClass::BOUND_AGGREGATE) {
+            return op;
+        }
+        const auto &bound_aggr = expr->Cast<BoundAggregateExpression>();
+        if (bound_aggr.IsDistinct() || bound_aggr.filter != nullptr || bound_aggr.order_bys != nullptr) {
+            return op;
+        }
+
+        if (bound_aggr.function.name == "count_star") {
+            input.emplace_back(COUNT_STAR_PROJ_IDX, *expr);
+            continue;
+        }
+
+        if (bound_aggr.children.size() != 1 ||
+            bound_aggr.children[0]->GetExpressionType() != ExpressionType::BOUND_COLUMN_REF) {
+            return op;
+        }
+        const auto &bound_col = bound_aggr.children[0]->Cast<BoundColumnRefExpression>();
+        const auto binding = Resolve(bound_col.binding, analyses, projections);
+        if (!binding || &binding->analysis.get != get) {
+            return op;
+        }
+        const TableColumnStorageIndex storage_index = binding->analysis.StorageIndex(binding->column_index);
+        input.emplace_back(storage_index, *expr);
+    }
+
+    if (!aggregate_pushdown(context, {*get, input})) {
+        return op;
+    }
+
+    // GET now returns one column per aggregate. Expand column into multiple
+    // column if there are many aggregates per column.
+    auto &column_ids = get->GetMutableColumnIds();
+    get->types.resize(N);
+    get->returned_types.resize(N);
+    column_ids.resize(N);
+
+    vector<string> names(N); // need a copy because we reference original names
+
+    for (idx_t i = 0; i < N; i++) {
+        const auto &[storage_index, expr] = input[i];
+        names[i] = storage_index == COUNT_STAR_PROJ_IDX ? "count_star()" : get->names[storage_index];
+        get->types[i] = expr.return_type;
+        get->returned_types[i] = expr.return_type;
+        column_ids[i] = ColumnIndex {i};
+    }
+    get->names = std::move(names);
+    get->projection_ids.clear();
+    get->table_index = agg.aggregate_index;
+
+    unique_ptr<LogicalOperator> &child = agg.children[0];
+    if (child->type == LOGICAL_GET) {
+        return std::move(child);
+    }
+    D_ASSERT(child->type == LOGICAL_PROJECTION);
+    D_ASSERT(child->children.size() == 1);
+    D_ASSERT(child->children[0]->type == LOGICAL_GET);
+    return std::move(child->children[0]);
+}
+
+LogicalGet *GetChildGet(const LogicalAggregate &agg) {
+    if (agg.children.size() != 1) {
+        return nullptr;
+    }
+    LogicalOperator &child = *agg.children[0];
+    LogicalOperator *op;
+    if (child.type == LOGICAL_GET) {
+        op = &child;
+    } else if (child.type == LOGICAL_PROJECTION && child.children.size() == 1 &&
+               child.children[0]->type == LOGICAL_GET) {
+        op = child.children[0].get();
+    } else {
+        return nullptr;
+    }
+    LogicalGet &get = op->Cast<LogicalGet>();
+    return get.function.bind == duckdb_vx_table_function_bind ? &get : nullptr;
+}
diff --git a/vortex-duckdb/cpp/expr.cpp b/vortex-duckdb/cpp/expr.cpp
index afe2573adc2..682999ab093 100644
--- a/vortex-duckdb/cpp/expr.cpp
+++ b/vortex-duckdb/cpp/expr.cpp
@@ -3,10 +3,12 @@
 
 #include "expr.h"
 #include "duckdb/function/scalar_function.hpp"
+#include "duckdb/function/aggregate_function.hpp"
 #include "duckdb/planner/expression/bound_between_expression.hpp"
 #include "duckdb/planner/expression/bound_columnref_expression.hpp"
 #include "duckdb/planner/expression/bound_comparison_expression.hpp"
 #include "duckdb/planner/expression/bound_constant_expression.hpp"
+#include "duckdb/planner/expression/bound_aggregate_expression.hpp"
 #include "duckdb/planner/expression/bound_function_expression.hpp"
 #include "duckdb/planner/expression/bound_operator_expression.hpp"
 #include "duckdb/planner/expression/bound_conjunction_expression.hpp"
@@ -21,6 +23,11 @@ extern "C" const char *duckdb_vx_sfunc_name(duckdb_vx_sfunc ffi_func) {
     return func->name.c_str();
 }
 
+extern "C" const char *duckdb_vx_agg_func_name(duckdb_vx_agg_func ffi) {
+    D_ASSERT(ffi);
+    return reinterpret_cast<AggregateFunction *>(ffi)->name.c_str();
+}
+
 extern "C" const char *duckdb_vx_expr_to_string(duckdb_vx_expr ffi_expr) {
     if (!ffi_expr) {
         return nullptr;
@@ -129,3 +136,9 @@ extern "C" void duckdb_vx_expr_get_bound_function(duckdb_vx_expr ffi_expr,
     out->scalar_function = reinterpret_cast<duckdb_vx_sfunc>(&expr.function);
     out->bind_info = expr.bind_info.get();
 }
+
+extern "C" duckdb_vx_agg_func duckdb_vx_expr_get_bound_aggregate_function(duckdb_vx_expr ffi_expr) {
+    D_ASSERT(ffi_expr);
+    auto &expr = reinterpret_cast<Expression *>(ffi_expr)->Cast<BoundAggregateExpression>();
+    return reinterpret_cast<duckdb_vx_agg_func>(&expr.function);
+}
diff --git a/vortex-duckdb/cpp/include/aggregate_fn_pushdown.hpp b/vortex-duckdb/cpp/include/aggregate_fn_pushdown.hpp
new file mode 100644
index 00000000000..ae3d95e69e7
--- /dev/null
+++ b/vortex-duckdb/cpp/include/aggregate_fn_pushdown.hpp
@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+#pragma once
+#include "scalar_fn_pushdown.hpp"
+#include "duckdb/optimizer/optimizer_extension.hpp"
+
+using namespace duckdb;
+
+// Push UNGROUPED_AGGREGATE's of form agg(T) and count_star() into GET.
+LogicalOperatorPtr TryPushdownAggregateFunctions(ClientContext &context, LogicalOperatorPtr plan);
+
+LogicalOperatorPtr RewriteAggregates(ClientContext &context,
+                                     LogicalOperatorPtr op,
+                                     Analyses &analyses,
+                                     const Projections &projections);
+
+LogicalOperatorPtr TryReplaceAggregate(ClientContext &context,
+                                       LogicalOperatorPtr op,
+                                       Analyses &analyses,
+                                       const Projections &projections);
+
+// return GET for UNGROUPED_AGGREGATE -> [GET] or for UNGROUPED_AGGREGATE ->
+// PROJECTION -> [GET], nullptr if not found.
+LogicalGet *GetChildGet(const LogicalAggregate &agg);
diff --git a/vortex-duckdb/cpp/include/expr.h b/vortex-duckdb/cpp/include/expr.h
index 5b7997596d6..5d8f100b89d 100644
--- a/vortex-duckdb/cpp/include/expr.h
+++ b/vortex-duckdb/cpp/include/expr.h
@@ -10,11 +10,14 @@ extern "C" {
 #endif
 
 typedef struct duckdb_vx_sfunc_ *duckdb_vx_sfunc;
+typedef struct duckdb_vx_agg_func_ *duckdb_vx_agg_func;
 
 const char *duckdb_vx_sfunc_name(duckdb_vx_sfunc ffi_func);
 
 typedef struct duckdb_vx_expr_ *duckdb_vx_expr;
 
+const char *duckdb_vx_agg_func_name(duckdb_vx_agg_func func);
+
 /// Return the string representation of the expression. Must be freed with `duckdb_free`.
 const char *duckdb_vx_expr_to_string(duckdb_vx_expr expr);
 
@@ -264,6 +267,8 @@ typedef struct {
 
 void duckdb_vx_expr_get_bound_function(duckdb_vx_expr expr, duckdb_vx_expr_bound_function *out);
 
+duckdb_vx_agg_func duckdb_vx_expr_get_bound_aggregate_function(duckdb_vx_expr expr);
+
 #ifdef __cplusplus /* End C ABI */
 }
 #endif
diff --git a/vortex-duckdb/cpp/include/table_function.h b/vortex-duckdb/cpp/include/table_function.h
index f59c0330300..e3567131a08 100644
--- a/vortex-duckdb/cpp/include/table_function.h
+++ b/vortex-duckdb/cpp/include/table_function.h
@@ -88,6 +88,10 @@ typedef struct {
 
 duckdb_state duckdb_vx_register_table_functions(duckdb_database ffi_db);
 
+typedef struct duckdb_vx_agg_input_ *duckdb_vx_agg_input;
+idx_t duckdb_vx_aggregate_len(duckdb_vx_agg_input ffi);
+duckdb_vx_expr duckdb_vx_aggregate_i(duckdb_vx_agg_input ffi, idx_t i, idx_t *proj_idx);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/vortex-duckdb/cpp/include/table_function.hpp b/vortex-duckdb/cpp/include/table_function.hpp
index 8cdb813f4a7..82a757568ae 100644
--- a/vortex-duckdb/cpp/include/table_function.hpp
+++ b/vortex-duckdb/cpp/include/table_function.hpp
@@ -25,3 +25,12 @@ struct TableFunctionProjectionExpressionInput {
 // true if we can push down the expression, false otherwise
 bool projection_expression_pushdown(duckdb::ClientContext &context,
                                     const TableFunctionProjectionExpressionInput &input);
+
+// API is subject to change, but you can push only none or all aggregates since
+// this changes output chunk cardinality
+struct TableFunctionUngroupedAggregateInput {
+    const duckdb::LogicalGet &get;
+    const duckdb::vector<std::pair<idx_t, const duckdb::Expression &>> &projections;
+};
+
+bool aggregate_pushdown(duckdb::ClientContext &context, const TableFunctionUngroupedAggregateInput &input);
diff --git a/vortex-duckdb/cpp/table_function.cpp b/vortex-duckdb/cpp/table_function.cpp
index 2b351593450..de5869a0497 100644
--- a/vortex-duckdb/cpp/table_function.cpp
+++ b/vortex-duckdb/cpp/table_function.cpp
@@ -4,6 +4,7 @@
 #include "data.hpp"
 #include "error.hpp"
 #include "table_function.hpp"
+#include "expr.h"
 #include "vortex_duckdb.h"
 #include "table_function.h"
 #include "vortex.h"
@@ -171,12 +172,16 @@ struct CTableBindResult {
     vector<string> &names;
 };
 
+// This is a flaw of Duckdb API which doesn't allow passing non-const
+// expressions. We never modify the value on Rust side.
+static duckdb_vx_expr get_ffi_expr(const Expression &expr) {
+    return reinterpret_cast<duckdb_vx_expr>(const_cast<Expression *>(&expr));
+}
+
 bool projection_expression_pushdown(ClientContext &, const TableFunctionProjectionExpressionInput &input) {
     const auto &bind = input.get.bind_data->Cast<CTableBindData>();
 
-    // This is a flaw of Duckdb API which doesn't allow passing non-const
-    // expressions. We never modify the value on Rust side.
-    auto ffi_expr = reinterpret_cast<duckdb_vx_expr>(const_cast<Expression *>(&input.expression));
+    duckdb_vx_expr ffi_expr = get_ffi_expr(input.expression);
     void *const ffi_bind = bind.ffi_data->DataPtr();
     duckdb_vx_error error_out = nullptr;
 
@@ -191,6 +196,33 @@ bool projection_expression_pushdown(ClientContext &, const TableFunctionProjecti
     return ret;
 }
 
+using Projections = vector<std::pair<idx_t, const Expression &>>;
+
+extern "C" {
+idx_t duckdb_vx_aggregate_len(duckdb_vx_agg_input ffi) {
+    return reinterpret_cast<const Projections *>(ffi)->size();
+}
+
+duckdb_vx_expr duckdb_vx_aggregate_i(duckdb_vx_agg_input ffi, idx_t i, idx_t *proj_idx) {
+    const Projections &projections = *reinterpret_cast<const Projections *>(ffi);
+    *proj_idx = projections[i].first;
+    return get_ffi_expr(projections[i].second);
+}
+}
+
+bool aggregate_pushdown(ClientContext &, const TableFunctionUngroupedAggregateInput &input) {
+    const auto &bind = input.get.bind_data->Cast<CTableBindData>();
+    void *const ffi_bind = bind.ffi_data->DataPtr();
+    duckdb_vx_error error_out = nullptr;
+    const auto ffi_input =
+        reinterpret_cast<duckdb_vx_agg_input>(const_cast<Projections *>(&input.projections));
+    const bool res = duckdb_table_function_pushdown_projection_aggregates(ffi_bind, ffi_input, &error_out);
+    if (error_out) {
+        throw BinderException(IntoErrString(error_out));
+    }
+    return res;
+}
+
 /**
  * Called for every new query. For example, if there is a VIEW over *.vortex,
  * and after a query another file is added matching the glob, for second query
@@ -238,10 +270,11 @@ unique_ptr<GlobalTableFunctionState> c_init_global(ClientContext &context, Table
 }
 
 unique_ptr<LocalTableFunctionState>
-init_local(ExecutionContext &, TableFunctionInitInput &, GlobalTableFunctionState *global_state) {
+init_local(ExecutionContext &, TableFunctionInitInput &input, GlobalTableFunctionState *global_state) {
+    const void *const ffi_bind = input.bind_data->Cast<CTableBindData>().ffi_data->DataPtr();
     void *const ffi_global = global_state->Cast<CTableGlobalData>().ffi_data->DataPtr();
 
-    duckdb_vx_data ffi_local_data = duckdb_table_function_init_local(ffi_global);
+    duckdb_vx_data ffi_local_data = duckdb_table_function_init_local(ffi_bind, ffi_global);
     auto cdata = unique_ptr<CData>(reinterpret_cast<CData *>(ffi_local_data));
     return make_uniq<CTableLocalData>(std::move(cdata));
 }
diff --git a/vortex-duckdb/cpp/vortex_duckdb.cpp b/vortex-duckdb/cpp/vortex_duckdb.cpp
index 091f98a1703..74670381861 100644
--- a/vortex-duckdb/cpp/vortex_duckdb.cpp
+++ b/vortex-duckdb/cpp/vortex_duckdb.cpp
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+#include "aggregate_fn_pushdown.hpp"
 #include "data.hpp"
 #include "error.hpp"
 #include "scalar_fn_pushdown.hpp"
@@ -269,6 +270,7 @@ extern "C" duckdb_blob duckdb_vx_value_get_geometry(duckdb_value value) {
 
 static void VortexOptimizeFunction(OptimizerExtensionInput &input, unique_ptr<LogicalOperator> &plan) {
     plan = TryPushdownScalarFunctions(input.context, std::move(plan));
+    plan = TryPushdownAggregateFunctions(input.context, std::move(plan));
 }
 
 struct VortexOptimizerExtension final : OptimizerExtension {
diff --git a/vortex-duckdb/include/vortex.h b/vortex-duckdb/include/vortex.h
index c63b20cbbb0..e8e4550e108 100644
--- a/vortex-duckdb/include/vortex.h
+++ b/vortex-duckdb/include/vortex.h
@@ -11,6 +11,8 @@
 
 #pragma once
 
+#define COUNT_STAR_PROJ_IDX UINT64_MAX
+
 #ifdef __cplusplus
 extern "C" {
 #endif // __cplusplus
@@ -40,6 +42,11 @@ bool duckdb_table_function_pushdown_projection_expression(void *bind_data,
                                                           size_t column_id,
                                                           duckdb_vx_error *error_out);
 
+extern
+bool duckdb_table_function_pushdown_projection_aggregates(void *bind_data,
+                                                          duckdb_vx_agg_input input,
+                                                          duckdb_vx_error *error_out);
+
 extern
 void duckdb_table_function_scan(void *global_init_data,
                                 void *local_init_data,
@@ -56,7 +63,9 @@ extern
 duckdb_vx_data duckdb_table_function_init_global(const duckdb_vx_tfunc_init_input *init_input,
                                                  duckdb_vx_error *error_out);
 
-extern duckdb_vx_data duckdb_table_function_init_local(void *global_init_data);
+extern
+duckdb_vx_data duckdb_table_function_init_local(const void *bind_data,
+                                                void *global_init_data);
 
 extern
 duckdb_vx_data duckdb_table_function_bind(duckdb_vx_tfunc_bind_input bind_input,
diff --git a/vortex-duckdb/src/convert/expr.rs b/vortex-duckdb/src/convert/expr.rs
index 387b644fe30..15e4d5696ce 100644
--- a/vortex-duckdb/src/convert/expr.rs
+++ b/vortex-duckdb/src/convert/expr.rs
@@ -4,6 +4,17 @@
 use std::sync::Arc;
 
 use tracing::debug;
+use vortex::aggregate_fn::Accumulator;
+use vortex::aggregate_fn::DynAccumulator;
+use vortex::aggregate_fn::EmptyOptions;
+use vortex::aggregate_fn::NumericalAggregateOpts;
+use vortex::aggregate_fn::combined::PairOptions;
+use vortex::aggregate_fn::fns::count::Count;
+use vortex::aggregate_fn::fns::first::First;
+use vortex::aggregate_fn::fns::max::Max;
+use vortex::aggregate_fn::fns::mean::Mean;
+use vortex::aggregate_fn::fns::min::Min;
+use vortex::aggregate_fn::fns::sum::Sum;
 use vortex::dtype::DType;
 use vortex::dtype::Nullability;
 use vortex::dtype::PType;
@@ -40,7 +51,6 @@ use vortex::scalar_fn::fns::operators::Operator;
 
 use crate::cpp::DUCKDB_TYPE;
 use crate::cpp::DUCKDB_VX_EXPR_TYPE;
-use crate::duckdb;
 use crate::duckdb::BoundFunction;
 use crate::duckdb::BoundOperator;
 use crate::duckdb::ExpressionClass;
@@ -50,9 +60,10 @@ use crate::duckdb::ExpressionClass::BoundComparison;
 use crate::duckdb::ExpressionClass::BoundConjunction;
 use crate::duckdb::ExpressionClass::BoundConstant;
 use crate::duckdb::ExpressionClass::BoundRef;
+use crate::duckdb::ExpressionRef;
 use crate::projection::DuckdbField;
 
-fn from_bound_str(value: &duckdb::ExpressionRef) -> VortexResult<String> {
+fn from_bound_str(value: &ExpressionRef) -> VortexResult<String> {
     match value.as_class().vortex_expect("unknown class") {
         BoundConstant(constant) => Ok(constant.value.as_string().as_str().to_owned()),
         _ => vortex_bail!("Expected string expression, got {:?}", value.as_class_id()),
@@ -60,7 +71,7 @@ fn from_bound_str(value: &duckdb::ExpressionRef) -> VortexResult<String> {
 }
 
 /// Whether the expression's return type is a `LIST` or fixed-size `ARRAY`.
-fn returns_a_list(expr: &duckdb::ExpressionRef) -> bool {
+fn returns_a_list(expr: &ExpressionRef) -> bool {
     matches!(
         expr.return_type().as_type_id(),
         DUCKDB_TYPE::DUCKDB_TYPE_LIST | DUCKDB_TYPE::DUCKDB_TYPE_ARRAY
@@ -171,14 +182,12 @@ fn try_from_bound_function(
     Ok(Some(expr))
 }
 
-pub fn try_from_bound_expression(
-    value: &duckdb::ExpressionRef,
-) -> VortexResult<Option<Expression>> {
+pub fn try_from_bound_expression(value: &ExpressionRef) -> VortexResult<Option<Expression>> {
     try_from_expression_inner(value, None)
 }
 
 pub(super) fn try_from_bound_expression_with_col_sub(
-    value: &duckdb::ExpressionRef,
+    value: &ExpressionRef,
     col_sub: &Expression,
 ) -> VortexResult<Option<Expression>> {
     try_from_expression_inner(value, Some(col_sub))
@@ -201,7 +210,7 @@ fn is_supported_length_alias(func: &BoundFunction) -> bool {
 // push it.
 // Example: optional filters may fail to parse on our side (we return
 // Ok(None)), so we don't allow pushing these.
-pub fn can_push_expression(value: &duckdb::ExpressionRef) -> bool {
+pub fn can_push_expression(value: &ExpressionRef) -> bool {
     let Some(value) = value.as_class() else {
         return false;
     };
@@ -241,6 +250,7 @@ pub fn can_push_expression(value: &duckdb::ExpressionRef) -> bool {
             }
             op.children().all(can_push_expression)
         }
+        ExpressionClass::BoundAggregate(_) => false,
     }
 }
 
@@ -252,7 +262,7 @@ fn list_length_on_field(field: &DuckdbField) -> Expression {
 }
 
 pub fn try_from_projection_expression(
-    value: &duckdb::ExpressionRef,
+    value: &ExpressionRef,
     field: &DuckdbField,
 ) -> VortexResult<Option<Expression>> {
     let Some(value) = value.as_class() else {
@@ -280,10 +290,61 @@ pub fn try_from_projection_expression(
     })
 }
 
+// Aggregations we handle in Vortex and not in Duckdb
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PushedAggregate {
+    Min,
+    Max,
+    Sum,
+    Mean,
+    // Also used for ANY_VALUE() which is allowed by definition
+    First,
+    // Valid values in column
+    Count,
+}
+
+impl PushedAggregate {
+    pub fn build(self, dtype: DType) -> VortexResult<Box<dyn DynAccumulator>> {
+        let opts = NumericalAggregateOpts::default();
+        Ok(match self {
+            Self::Min => Box::new(Accumulator::try_new(Min, opts, dtype)?),
+            Self::Max => Box::new(Accumulator::try_new(Max, opts, dtype)?),
+            Self::Sum => Box::new(Accumulator::try_new(Sum, opts, dtype)?),
+            Self::Mean => Box::new(Accumulator::try_new(
+                Mean::combined(),
+                PairOptions(opts, opts),
+                dtype,
+            )?),
+            Self::First => Box::new(Accumulator::try_new(First, EmptyOptions, dtype)?),
+            Self::Count => Box::new(Accumulator::try_new(Count, opts, dtype)?),
+        })
+    }
+}
+
+pub fn try_from_projection_aggregate(
+    expr: &ExpressionRef,
+) -> VortexResult<Option<PushedAggregate>> {
+    let Some(expr) = expr.as_class() else {
+        return Ok(None);
+    };
+    let ExpressionClass::BoundAggregate(agg) = expr else {
+        return Ok(None);
+    };
+    Ok(Some(match agg.aggregate_function.name() {
+        "min" => PushedAggregate::Min,
+        "max" => PushedAggregate::Max,
+        "sum" | "sum_no_overflow" => PushedAggregate::Sum,
+        "avg" | "mean" => PushedAggregate::Mean,
+        "first" | "any_value" => PushedAggregate::First,
+        "count" => PushedAggregate::Count,
+        _ => return Ok(None),
+    }))
+}
+
 // If you want to add support for other expressions, also change
 // can_push_expression
 fn try_from_expression_inner(
-    value: &duckdb::ExpressionRef,
+    value: &ExpressionRef,
     col_sub: Option<&Expression>,
 ) -> VortexResult<Option<Expression>> {
     let Some(value) = value.as_class() else {
@@ -390,6 +451,7 @@ fn try_from_expression_inner(
                 _ => vortex_bail!("unexpected operator {:?} in bound conjunction", conj.op),
             }
         }
+        ExpressionClass::BoundAggregate(_) => return Ok(None),
     }))
 }
 
diff --git a/vortex-duckdb/src/convert/mod.rs b/vortex-duckdb/src/convert/mod.rs
index 0d641a73457..f1b5ba5bb10 100644
--- a/vortex-duckdb/src/convert/mod.rs
+++ b/vortex-duckdb/src/convert/mod.rs
@@ -8,8 +8,10 @@ mod table_filter;
 mod vector;
 
 pub use dtype::FromLogicalType;
+pub use expr::PushedAggregate;
 pub use expr::can_push_expression;
 pub use expr::try_from_bound_expression;
+pub use expr::try_from_projection_aggregate;
 pub use expr::try_from_projection_expression;
 pub use scalar::*;
 pub use table_filter::try_from_table_filter;
diff --git a/vortex-duckdb/src/duckdb/aggregate_pushdown.rs b/vortex-duckdb/src/duckdb/aggregate_pushdown.rs
new file mode 100644
index 00000000000..edbc486baa6
--- /dev/null
+++ b/vortex-duckdb/src/duckdb/aggregate_pushdown.rs
@@ -0,0 +1,33 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use crate::cpp;
+use crate::duckdb::Expression;
+use crate::duckdb::ExpressionRef;
+use crate::lifetime_wrapper;
+
+lifetime_wrapper!(AggregatePushdownInput, cpp::duckdb_vx_agg_input, |_| {});
+
+pub struct AggregateExpression<'a> {
+    pub expr: &'a ExpressionRef,
+    pub projection_id: u64,
+}
+
+impl AggregatePushdownInputRef {
+    pub fn len(&self) -> usize {
+        let len = unsafe { cpp::duckdb_vx_aggregate_len(self.as_ptr()) };
+        len as usize
+    }
+
+    pub fn get(&'_ self, index: usize) -> AggregateExpression<'_> {
+        let mut projection_id = 0u64;
+        let expr = unsafe {
+            cpp::duckdb_vx_aggregate_i(self.as_ptr(), index as u64, &mut projection_id as *mut u64)
+        };
+        let expr = unsafe { Expression::borrow(expr) };
+        AggregateExpression {
+            expr,
+            projection_id,
+        }
+    }
+}
diff --git a/vortex-duckdb/src/duckdb/expr.rs b/vortex-duckdb/src/duckdb/expr.rs
index 48f255f744e..06756a2b928 100644
--- a/vortex-duckdb/src/duckdb/expr.rs
+++ b/vortex-duckdb/src/duckdb/expr.rs
@@ -2,13 +2,14 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::ffi::CStr;
-use std::ffi::c_void;
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::ptr;
 
 use crate::cpp;
 use crate::cpp::duckdb_vx_expr_class;
+use crate::duckdb::AggregateFunction;
+use crate::duckdb::AggregateFunctionRef;
 use crate::duckdb::DDBString;
 use crate::duckdb::LogicalType;
 use crate::duckdb::LogicalTypeRef;
@@ -143,9 +144,15 @@ impl ExpressionRef {
                     ExpressionClass::BoundFunction(BoundFunction {
                         children,
                         scalar_function: unsafe { ScalarFunction::borrow(out.scalar_function) },
-                        bind_info: out.bind_info,
                     })
                 }
+                cpp::DUCKDB_VX_EXPR_CLASS::DUCKDB_VX_EXPR_CLASS_BOUND_AGGREGATE => {
+                    let aggregate_function = unsafe {
+                        let ptr = cpp::duckdb_vx_expr_get_bound_aggregate_function(self.as_ptr());
+                        AggregateFunction::borrow(ptr)
+                    };
+                    ExpressionClass::BoundAggregate(BoundAggregate { aggregate_function })
+                }
                 cpp::DUCKDB_VX_EXPR_CLASS::DUCKDB_VX_EXPR_CLASS_BOUND_REF => {
                     ExpressionClass::BoundRef
                 }
@@ -165,6 +172,7 @@ pub enum ExpressionClass<'a> {
     BoundBetween(BoundBetween<'a>),
     BoundOperator(BoundOperator<'a>),
     BoundFunction(BoundFunction<'a>),
+    BoundAggregate(BoundAggregate<'a>),
     /// Column inside ExpressionFilter for expression pushed down to Vortex.
     BoundRef,
 }
@@ -173,6 +181,10 @@ pub struct BoundColumnRef {
     pub name: DDBString,
 }
 
+pub struct BoundAggregate<'a> {
+    pub aggregate_function: &'a AggregateFunctionRef,
+}
+
 pub struct BoundConstant<'a> {
     pub value: &'a ValueRef,
 }
@@ -222,7 +234,6 @@ impl<'a> BoundOperator<'a> {
 pub struct BoundFunction<'a> {
     children: &'a [cpp::duckdb_vx_expr],
     pub scalar_function: &'a ScalarFunctionRef,
-    pub bind_info: *const c_void,
 }
 
 impl<'a> BoundFunction<'a> {
diff --git a/vortex-duckdb/src/duckdb/scalar_function.rs b/vortex-duckdb/src/duckdb/function.rs
similarity index 58%
rename from vortex-duckdb/src/duckdb/scalar_function.rs
rename to vortex-duckdb/src/duckdb/function.rs
index 175b0e70d5c..0441fe43548 100644
--- a/vortex-duckdb/src/duckdb/scalar_function.rs
+++ b/vortex-duckdb/src/duckdb/function.rs
@@ -8,6 +8,7 @@ use crate::cpp;
 use crate::lifetime_wrapper;
 
 lifetime_wrapper!(ScalarFunction, cpp::duckdb_vx_sfunc, |_| {});
+lifetime_wrapper!(AggregateFunction, cpp::duckdb_vx_agg_func, |_| {});
 
 impl ScalarFunctionRef {
     pub fn name(&self) -> &str {
@@ -20,3 +21,15 @@ impl ScalarFunctionRef {
         }
     }
 }
+
+impl AggregateFunctionRef {
+    pub fn name(&self) -> &str {
+        unsafe {
+            let name_ptr = cpp::duckdb_vx_agg_func_name(self.as_ptr());
+            std::ffi::CStr::from_ptr(name_ptr)
+                .to_str()
+                .map_err(|e| vortex_err!("invalid utf-8: {e}"))
+                .vortex_expect("aggregate function name should be valid UTF-8")
+        }
+    }
+}
diff --git a/vortex-duckdb/src/duckdb/mod.rs b/vortex-duckdb/src/duckdb/mod.rs
index 27ea994b1e7..95dc8adc186 100644
--- a/vortex-duckdb/src/duckdb/mod.rs
+++ b/vortex-duckdb/src/duckdb/mod.rs
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+mod aggregate_pushdown;
 mod bind_input;
 mod connection;
 mod data;
@@ -8,11 +9,11 @@ mod data_chunk;
 mod database;
 mod ddb_string;
 mod expr;
+mod function;
 mod logical_type;
 mod macro_;
 mod query_result;
 mod reusable_dict;
-mod scalar_function;
 mod selection_vector;
 mod string_map;
 mod table_filter;
@@ -24,6 +25,7 @@ mod vector_buffer;
 use std::ffi::c_void;
 use std::ptr;
 
+pub use aggregate_pushdown::*;
 pub use bind_input::*;
 pub use connection::*;
 pub use data::*;
@@ -31,10 +33,10 @@ pub use data_chunk::*;
 pub use database::*;
 pub use ddb_string::*;
 pub use expr::*;
+pub use function::*;
 pub use logical_type::*;
 pub use query_result::*;
 pub use reusable_dict::*;
-pub use scalar_function::*;
 pub use selection_vector::*;
 pub use string_map::*;
 pub use table_filter::*;
diff --git a/vortex-duckdb/src/ffi.rs b/vortex-duckdb/src/ffi.rs
index 880c3f7f01c..e67abf9423a 100644
--- a/vortex-duckdb/src/ffi.rs
+++ b/vortex-duckdb/src/ffi.rs
@@ -17,6 +17,7 @@ use crate::copy::copy_to_finalize;
 use crate::copy::copy_to_initialize_global;
 use crate::copy::copy_to_sink;
 use crate::cpp;
+use crate::duckdb::AggregatePushdownInput;
 use crate::duckdb::BindInput;
 use crate::duckdb::BindResult;
 use crate::duckdb::Data;
@@ -38,6 +39,7 @@ use crate::table_function::get_partition_data;
 use crate::table_function::init_global;
 use crate::table_function::init_local;
 use crate::table_function::pushdown_complex_filter;
+use crate::table_function::pushdown_projection_aggregates;
 use crate::table_function::pushdown_projection_expression;
 use crate::table_function::scan;
 use crate::table_function::statistics;
@@ -126,6 +128,20 @@ unsafe extern "C-unwind" fn duckdb_table_function_pushdown_projection_expression
     })
 }
 
+#[unsafe(no_mangle)]
+pub unsafe extern "C-unwind" fn duckdb_table_function_pushdown_projection_aggregates(
+    bind_data: *mut c_void,
+    input: cpp::duckdb_vx_agg_input,
+    error_out: *mut cpp::duckdb_vx_error,
+) -> bool {
+    let bind_data = unsafe { bind_data.cast::<TableFunctionBind>().as_mut() }
+        .vortex_expect("bind_data null pointer");
+    let input = unsafe { AggregatePushdownInput::borrow(input) };
+    try_or(error_out, || {
+        pushdown_projection_aggregates(bind_data, &input)
+    })
+}
+
 #[unsafe(no_mangle)]
 unsafe extern "C-unwind" fn duckdb_table_function_scan(
     global_init_data: *mut c_void,
@@ -207,12 +223,15 @@ pub unsafe extern "C-unwind" fn duckdb_table_function_init_global(
 
 #[unsafe(no_mangle)]
 pub unsafe extern "C-unwind" fn duckdb_table_function_init_local(
+    bind_data: *const c_void,
     global_init_data: *mut c_void,
 ) -> cpp::duckdb_vx_data {
+    let bind_data = unsafe { bind_data.cast::<TableFunctionBind>().as_ref() }
+        .vortex_expect("bind_data null pointer");
     let global_init_data = unsafe { global_init_data.cast::<TableFunctionGlobal>().as_ref() }
         .vortex_expect("global_init_data null pointer");
 
-    let init_data = init_local(global_init_data);
+    let init_data = init_local(bind_data, global_init_data);
     Data::from(Box::new(init_data)).as_ptr()
 }
 
diff --git a/vortex-duckdb/src/projection.rs b/vortex-duckdb/src/projection.rs
index 4521115666c..3bc47f8c679 100644
--- a/vortex-duckdb/src/projection.rs
+++ b/vortex-duckdb/src/projection.rs
@@ -23,6 +23,7 @@ use crate::convert::try_from_virtual_column_filter;
 use crate::duckdb::LogicalType;
 use crate::duckdb::TableFilterClass;
 use crate::duckdb::TableFilterSetRef;
+use crate::table_function::ColumnAggregate;
 
 // See MultiFileReader for constants
 
@@ -187,6 +188,27 @@ impl Projection {
             file_row_number_column_pos,
         }
     }
+
+    pub fn new_aggregate(aggregates: &[ColumnAggregate], fields: &[DuckdbField]) -> Self {
+        let mut names = Vec::with_capacity(aggregates.len());
+        let mut seen: Vec<u64> = Vec::with_capacity(aggregates.len());
+        for aggregate in aggregates {
+            let ColumnAggregate::Real { projection_id, .. } = aggregate else {
+                continue;
+            };
+            if seen.contains(projection_id) {
+                continue;
+            }
+            seen.push(*projection_id);
+            let projection_id: usize = projection_id.as_();
+            names.push(fields[projection_id].name.as_str());
+        }
+        Projection {
+            projection: select(names, root()),
+            file_index_column_pos: None,
+            file_row_number_column_pos: None,
+        }
+    }
 }
 
 pub struct Filter {
diff --git a/vortex-duckdb/src/table_function.rs b/vortex-duckdb/src/table_function.rs
index 5151e47a464..4b8dccdcf3c 100644
--- a/vortex-duckdb/src/table_function.rs
+++ b/vortex-duckdb/src/table_function.rs
@@ -13,15 +13,19 @@ use custom_labels::CURRENT_LABELSET;
 use futures::StreamExt;
 use itertools::Itertools;
 use num_traits::AsPrimitive;
+use parking_lot::Mutex;
 use static_assertions::assert_impl_all;
 use tracing::debug;
+use vortex::aggregate_fn::DynAccumulator;
 use vortex::array::ArrayRef;
 use vortex::array::Canonical;
+use vortex::array::ExecutionCtx;
 use vortex::array::VortexSessionExecute as _;
 use vortex::array::arrays::ScalarFn;
 use vortex::array::arrays::Struct;
 use vortex::array::arrays::StructArray;
 use vortex::array::arrays::scalar_fn::ScalarFnArrayExt;
+use vortex::array::arrays::struct_::StructArrayExt;
 use vortex::array::optimizer::ArrayOptimizer;
 use vortex::error::VortexExpect;
 use vortex::error::VortexResult;
@@ -39,14 +43,19 @@ use vortex::scalar_fn::fns::operators::Operator;
 use vortex::scalar_fn::fns::pack::Pack;
 use vortex::scan::DataSource;
 use vortex::scan::ScanRequest;
+use vortex_utils::aliases::hash_map::HashMap;
 use vortex_utils::parallelism::get_available_parallelism;
 
 use crate::RUNTIME;
 use crate::SESSION;
 use crate::column_statistics::ColumnStatistics;
 use crate::column_statistics::ColumnStatisticsAggregate;
+use crate::convert::PushedAggregate;
 use crate::convert::try_from_bound_expression;
+use crate::convert::try_from_projection_aggregate;
 use crate::convert::try_from_projection_expression;
+use crate::duckdb::AggregateExpression;
+use crate::duckdb::AggregatePushdownInputRef;
 use crate::duckdb::BindInputRef;
 use crate::duckdb::BindResultRef;
 use crate::duckdb::DataChunkRef;
@@ -62,6 +71,9 @@ use crate::projection::Filter;
 use crate::projection::Projection;
 use crate::projection::extract_schema_from_dtype;
 
+// see cpp/aggregate_fn_pushdown.cpp
+pub const COUNT_STAR_PROJ_IDX: u64 = u64::MAX;
+
 pub struct TableFunctionBind {
     data_source: Arc<MultiLayoutDataSource>,
     filter_exprs: Vec<Expression>,
@@ -69,6 +81,8 @@ pub struct TableFunctionBind {
     // There exists at least one non-optional table filter or at least one
     // complex filter is pushed down.
     has_non_optional_filter: AtomicBool,
+    // Non-empty iff this scan is aggregate
+    aggregates: Vec<ColumnAggregate>,
 }
 assert_impl_all!(TableFunctionBind: Send, Clone);
 
@@ -82,6 +96,7 @@ impl Clone for TableFunctionBind {
             has_non_optional_filter: AtomicBool::new(
                 self.has_non_optional_filter.load(Ordering::Relaxed),
             ),
+            aggregates: self.aggregates.clone(),
         }
     }
 }
@@ -117,6 +132,16 @@ pub struct TableFunctionGlobal {
     bytes_read: AtomicU64,
     file_index_column_pos: Option<usize>,
     file_row_number_column_pos: Option<usize>,
+
+    // Following fields are initialized and used only in aggregate scans
+    /// ArrayRef's scanned but not aggregated in "partials".
+    /// 0 means all arrays have been aggregated but output is not written.
+    /// u64::MAX means arrays have been aggregated and we've written output row
+    pending: Arc<AtomicU64>,
+    aggregates: Vec<ColumnAggregate>,
+    // Accumulated partials
+    partials: Mutex<Vec<Box<dyn DynAccumulator>>>,
+    row_count: AtomicU64,
 }
 assert_impl_all!(TableFunctionGlobal: Send, Sync);
 
@@ -126,6 +151,8 @@ pub struct TableFunctionLocal {
     exporter: Option<ArrayExporter>,
     partition_index: u64,
     file_index: usize,
+    // Aggregate scan accumulated partials. Empty for non-aggregate scan
+    partials: Vec<Box<dyn DynAccumulator>>,
 }
 
 pub struct PartitionData {
@@ -134,6 +161,15 @@ pub struct PartitionData {
     pub file_index: usize,
 }
 
+#[derive(Clone)]
+pub(crate) enum ColumnAggregate {
+    Real {
+        projection_id: u64,
+        aggregate: PushedAggregate,
+    },
+    CountStar,
+}
+
 #[derive(Debug)]
 pub enum Cardinality {
     /// Unknown number of rows
@@ -155,6 +191,7 @@ pub fn bind(input: &BindInputRef, result: &mut BindResultRef) -> VortexResult<Ta
         filter_exprs: vec![],
         column_fields,
         has_non_optional_filter: AtomicBool::new(false),
+        aggregates: vec![],
     })
 }
 
@@ -169,7 +206,11 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
         projection,
         file_index_column_pos,
         file_row_number_column_pos,
-    } = Projection::new(projection_ids, column_ids, &bind_data.column_fields);
+    } = if bind_data.aggregates.is_empty() {
+        Projection::new(projection_ids, column_ids, &bind_data.column_fields)
+    } else {
+        Projection::new_aggregate(&bind_data.aggregates, &bind_data.column_fields)
+    };
 
     let Filter {
         filter,
@@ -224,6 +265,9 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
     // available array chunk regardless of which partition it came from.
     let (tx, rx) = kanal::bounded_async(num_workers * 2);
 
+    let pending = Arc::new(AtomicU64::new(0));
+    let pending_producer = Arc::clone(&pending);
+
     // We drive one partition per worker thread. Each partition is driven as a spawned task
     // that pushes array chunks into the shared channel as they are produced. This spawning
     // allows all worker threads to drive the polling of all partitions, and then return the
@@ -232,6 +276,7 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
         .partitions()
         .map(move |partition| {
             let tx = tx.clone();
+            let pending = Arc::clone(&pending_producer);
             RUNTIME.handle().spawn(async move {
                 let partition = match partition {
                     Ok(partition) => partition,
@@ -254,6 +299,7 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
                     }
                 };
                 while let Some(item) = stream.next().await {
+                    pending.fetch_add(1, Ordering::Release);
                     if tx
                         .send(item.map(|a| (a, Arc::clone(&cache))))
                         .await
@@ -273,6 +319,9 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
 
     let iterator = RUNTIME.block_on_stream_thread_safe(|_handle| rx.into_stream());
 
+    let aggregates = bind_data.aggregates.clone();
+    let partials = build_partials(&aggregates, &bind_data.column_fields)?;
+
     Ok(TableFunctionGlobal {
         iterator,
         batch_id: AtomicU64::new(0),
@@ -280,10 +329,36 @@ pub fn init_global(init_input: &TableInitInput) -> VortexResult<TableFunctionGlo
         bytes_read: AtomicU64::new(0),
         file_index_column_pos,
         file_row_number_column_pos,
+        pending,
+        aggregates,
+        partials: Mutex::new(partials),
+        row_count: AtomicU64::new(0),
     })
 }
 
-pub fn init_local(global: &TableFunctionGlobal) -> TableFunctionLocal {
+fn build_partials(
+    aggregates: &[ColumnAggregate],
+    fields: &[DuckdbField],
+) -> VortexResult<Vec<Box<dyn DynAccumulator>>> {
+    aggregates
+        .iter()
+        .filter_map(|spec| match spec {
+            ColumnAggregate::Real {
+                projection_id,
+                aggregate,
+            } => {
+                let projection_id: usize = projection_id.as_();
+                Some(aggregate.build(fields[projection_id].dtype.clone()))
+            }
+            ColumnAggregate::CountStar => None,
+        })
+        .collect()
+}
+
+pub fn init_local(
+    bind_data: &TableFunctionBind,
+    global: &TableFunctionGlobal,
+) -> TableFunctionLocal {
     unsafe {
         use custom_labels::sys;
 
@@ -299,11 +374,109 @@ pub fn init_local(global: &TableFunctionGlobal) -> TableFunctionLocal {
         CURRENT_LABELSET.set(key, value);
     }
 
+    let partials = build_partials(&global.aggregates, &bind_data.column_fields)
+        // if aggregate initialization produced an error, it would error in
+        // init_global, see "partials" intialization there
+        .vortex_expect("local state aggregate initialization failed");
+
     TableFunctionLocal {
         iterator: global.iterator.clone(),
         exporter: None,
         partition_index: 0,
         file_index: 0,
+        partials,
+    }
+}
+
+fn convert_result(array: ArrayRef, mut ctx: &mut ExecutionCtx) -> VortexResult<StructArray> {
+    let array_result = array.optimize_recursive(ctx.session())?;
+    Ok(if let Some(array) = array_result.as_opt::<Struct>() {
+        array.into_owned()
+    } else if let Some(array) = array_result.as_opt::<ScalarFn>()
+        && let Some(pack_options) = array.scalar_fn().as_opt::<Pack>()
+    {
+        StructArray::new(
+            pack_options.names.clone(),
+            array.children(),
+            array.len(),
+            pack_options.nullability.into(),
+        )
+    } else {
+        array_result.execute::<Canonical>(&mut ctx)?.into_struct()
+    })
+}
+
+fn scan_aggregate(
+    local_state: &mut TableFunctionLocal,
+    global_state: &TableFunctionGlobal,
+    chunk: &mut DataChunkRef,
+) -> VortexResult<()> {
+    let aggregates_len = global_state.aggregates.len();
+    // seen[k] = output column for requested column k.
+    // If min(x), max(x), avg(y) are requested, seen = { 0: 0, 1: 1}
+    let mut seen: HashMap<u64, usize> = HashMap::with_capacity(aggregates_len);
+    // positions[k] = column id for accumulator k
+    // If min(x), max(x), avg(y) are requested, positions = [0, 0, 1]
+    let mut positions: Vec<usize> = Vec::with_capacity(aggregates_len);
+
+    for aggregate in &global_state.aggregates {
+        let ColumnAggregate::Real { projection_id, .. } = aggregate else {
+            continue;
+        };
+        let len = seen.len();
+        let pos = seen.entry_ref(projection_id).or_insert(len);
+        positions.push(*pos);
+    }
+    let has_count_star = local_state.partials.len() < aggregates_len;
+
+    let mut ctx = SESSION.create_execution_ctx();
+    loop {
+        let Some(result) = local_state.iterator.next() else {
+            // 0 means we're the last thread, u64::MAX means output is written.
+            // is_err() means CAS didn't succeed
+            if global_state
+                .pending
+                .compare_exchange(0, u64::MAX, Ordering::AcqRel, Ordering::Acquire)
+                .is_err()
+            {
+                return Ok(());
+            }
+
+            let mut accumulators = global_state.partials.lock();
+            let row_count = global_state.row_count.load(Ordering::Acquire) as i64;
+            let mut accum_iter = accumulators.iter_mut();
+            for (i, aggregate) in global_state.aggregates.iter().enumerate() {
+                let value = match aggregate {
+                    ColumnAggregate::Real { .. } => {
+                        let accum = accum_iter.next().vortex_expect("partial for real agg");
+                        Value::try_from(accum.finish()?)?
+                    }
+                    ColumnAggregate::CountStar => Value::from(row_count),
+                };
+                chunk.get_vector_mut(i).reference_value(&value);
+            }
+            chunk.set_len(1);
+            return Ok(());
+        };
+        let array = convert_result(result?.0, &mut ctx)?;
+
+        for (i, partial) in positions.iter().zip(local_state.partials.iter_mut()) {
+            partial.accumulate(array.unmasked_field(*i), &mut ctx)?;
+        }
+
+        {
+            let mut partials = global_state.partials.lock();
+            for (global, local) in partials.iter_mut().zip(&mut local_state.partials) {
+                global.combine_partials(local.flush()?)?;
+            }
+        }
+
+        if has_count_star {
+            global_state
+                .row_count
+                .fetch_add(array.len() as u64, Ordering::Relaxed);
+        }
+        global_state.pending.fetch_sub(1, Ordering::Release);
     }
 }
 
@@ -312,6 +485,10 @@ pub fn scan(
     global_state: &TableFunctionGlobal,
     chunk: &mut DataChunkRef,
 ) -> VortexResult<()> {
+    if !local_state.partials.is_empty() {
+        return scan_aggregate(local_state, global_state, chunk);
+    }
+
     loop {
         if local_state.exporter.is_none() {
             let mut ctx = SESSION.create_execution_ctx();
@@ -319,23 +496,8 @@ pub fn scan(
                 return Ok(());
             };
             let (array_result, conversion_cache) = result?;
-            let array_result = array_result.optimize_recursive(ctx.session())?;
             local_state.file_index = conversion_cache.file_index;
-
-            let array_result: StructArray = if let Some(array) = array_result.as_opt::<Struct>() {
-                array.into_owned()
-            } else if let Some(array) = array_result.as_opt::<ScalarFn>()
-                && let Some(pack_options) = array.scalar_fn().as_opt::<Pack>()
-            {
-                StructArray::new(
-                    pack_options.names.clone(),
-                    array.children(),
-                    array.len(),
-                    pack_options.nullability.into(),
-                )
-            } else {
-                array_result.execute::<Canonical>(&mut ctx)?.into_struct()
-            };
+            let array_result = convert_result(array_result, &mut ctx)?;
 
             local_state.exporter = Some(ArrayExporter::try_new(
                 &array_result,
@@ -447,6 +609,46 @@ pub fn pushdown_projection_expression(
     }
 }
 
+/// Turn a scan into an aggregate scan. Input is N aggregations, possibly over
+/// same columns. If we return true, optimized pass expands output to N columns,
+/// e.g. min(x), max(x) turns into min(x0), max(x1), 2 columns in output.
+pub fn pushdown_projection_aggregates(
+    bind_data: &mut TableFunctionBind,
+    input: &AggregatePushdownInputRef,
+) -> VortexResult<bool> {
+    let len = input.len();
+    let mut aggregates = Vec::with_capacity(len);
+    let mut has_non_count_star = false;
+
+    debug!(%len, "pushing down projection aggregates");
+    for i in 0..len {
+        let AggregateExpression {
+            expr,
+            projection_id,
+        } = input.get(i);
+        if projection_id == COUNT_STAR_PROJ_IDX {
+            aggregates.push(ColumnAggregate::CountStar);
+            continue;
+        }
+        let Some(aggregate) = try_from_projection_aggregate(expr)? else {
+            debug!(%expr, %i, "failed to push down projection aggregate");
+            return Ok(false);
+        };
+        debug!(%expr, %projection_id, %i, "pushed down projection aggregate");
+        aggregates.push(ColumnAggregate::Real {
+            projection_id,
+            aggregate,
+        });
+        has_non_count_star = true;
+    }
+    // DuckDB computes just count(*) faster than us
+    if !has_non_count_star {
+        return Ok(false);
+    }
+    bind_data.aggregates = aggregates;
+    Ok(true)
+}
+
 /// Get column-wise statistics. Available only if we're reading a single file.
 pub fn statistics(bind_data: &TableFunctionBind, column_index: usize) -> Option<ColumnStatistics> {
     let children = bind_data.data_source.children();
@@ -464,8 +666,16 @@ pub fn statistics(bind_data: &TableFunctionBind, column_index: usize) -> Option<
         Some(inner) => inner.file_stats().stats_sets(),
         None => return None,
     };
-    let stats_aggregate = ColumnStatisticsAggregate::new(&stats_sets[column_index]);
-    let dtype = bind_data.column_fields[column_index].dtype.clone();
+    let source_id = if bind_data.aggregates.is_empty() {
+        column_index
+    } else {
+        match bind_data.aggregates[column_index] {
+            ColumnAggregate::Real { projection_id, .. } => projection_id.as_(),
+            ColumnAggregate::CountStar => return None,
+        }
+    };
+    let dtype = bind_data.column_fields[source_id].dtype.clone();
+    let stats_aggregate = ColumnStatisticsAggregate::new(&stats_sets[source_id]);
     Some(ColumnStatistics::from(&stats_aggregate, dtype))
 }
 
diff --git a/vortex-sqllogictest/Cargo.toml b/vortex-sqllogictest/Cargo.toml
index ddfd79ef99e..e722ec8a7e8 100644
--- a/vortex-sqllogictest/Cargo.toml
+++ b/vortex-sqllogictest/Cargo.toml
@@ -18,7 +18,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 bigdecimal = { workspace = true }
 datafusion = { workspace = true }
-datafusion-functions-nested = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
 datafusion-sqllogictest = { workspace = true }
 indicatif = { workspace = true }
 regex = { workspace = true }
@@ -29,6 +29,7 @@ tokio = { workspace = true, features = ["full"] }
 vortex = { workspace = true, features = ["tokio"] }
 vortex-datafusion = { workspace = true }
 vortex-duckdb = { workspace = true }
+datafusion-functions-nested.workspace = true
 
 [lints]
 workspace = true
diff --git a/vortex-sqllogictest/bin/sqllogictests-runner.rs b/vortex-sqllogictest/bin/sqllogictests-runner.rs
index 80335ac8951..fd5aa0cec60 100644
--- a/vortex-sqllogictest/bin/sqllogictests-runner.rs
+++ b/vortex-sqllogictest/bin/sqllogictests-runner.rs
@@ -198,6 +198,11 @@ fn complete_files(
 }
 
 fn main() -> anyhow::Result<ExitCode> {
+    drop(
+        tracing_subscriber::fmt()
+            .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+            .try_init(),
+    );
     let mut raw_args: Vec<String> = std::env::args().collect();
     // We remove the `--complete` flag that isn't standard before we pass the rest.
     let complete = {
diff --git a/vortex-sqllogictest/slt/duckdb/aggregate_pushdown.slt b/vortex-sqllogictest/slt/duckdb/aggregate_pushdown.slt
new file mode 100644
index 00000000000..2bf91d7f007
--- /dev/null
+++ b/vortex-sqllogictest/slt/duckdb/aggregate_pushdown.slt
@@ -0,0 +1,66 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+include ../setup.slt.no
+
+statement ok
+COPY (
+WITH cte AS (SELECT generate_series AS i FROM generate_series(100000))
+SELECT (CASE WHEN i > 0 THEN i ELSE NULL END) AS i FROM cte
+)
+TO '${WORK_DIR}/agg-pushdown.vortex';
+
+query TT
+EXPLAIN
+SELECT sum(i), min(i), max(i), avg(i), mean(i), any_value(i), first(i), count(*), count(i)
+FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+<!REGEX>:.*UNGROUPED_AGGREGATE.*
+
+query I
+SELECT sum(i) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+5000050000
+
+query II
+SELECT min(i), max(i) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+1 100000
+
+query RR
+SELECT avg(i), mean(i) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+50000.5 50000.5
+
+query II
+SELECT count(*), count(i) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+100001 100000
+
+# inverse order to test count(*) not being first
+query II
+SELECT count(i), count(*) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+100000 100001
+
+# Just count() is rejected because not pushing it down is faster:
+# we just set chunk length and don't do any aggregation ourselves
+query TT
+EXPLAIN SELECT count(*) FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+<REGEX>:.*UNGROUPED_AGGREGATE.*
+
+query TT
+EXPLAIN SELECT count(), count(), count() FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+<REGEX>:.*UNGROUPED_AGGREGATE.*
+
+query III
+SELECT count(), count(), count() FROM '${WORK_DIR}/agg-pushdown.vortex';
+----
+100001 100001 100001
+
+# aggregate over scalar i.e. SELECT mean(strlen(str))
+# + cte
+# + view + recursive view
+# + cte referencing any_value()