Apache Arrow AceroExperimental

# pip install pyarrow==26.0.0
import pyarrow as pa
import pyarrow.compute as pc
from pyarrow.acero import (Declaration, TableSourceNodeOptions, FilterNodeOptions,
                           HashJoinNodeOptions, AggregateNodeOptions, OrderByNodeOptions)

orders = pa.table({"order_id": [1, 2, 3, 4, 5], "customer_id": [10, 20, 10, 30, 20],
                   "amount": [25.0, 40.0, 15.0, 90.0, 5.0]})
customers = pa.table({"customer_id": [10, 20, 30], "country": ["DE", "US", "DE"]})

# Acero has no SQL parser or optimizer: you spell out every node of the plan
plan = Declaration.from_sequence([
    Declaration("table_source", TableSourceNodeOptions(orders)),
    Declaration("filter", FilterNodeOptions(pc.field("amount") > 10)),
])
plan = Declaration("hashjoin", HashJoinNodeOptions("inner", ["customer_id"], ["customer_id"]),
                   inputs=[plan, Declaration("table_source", TableSourceNodeOptions(customers))])
plan = Declaration.from_sequence([
    plan,
    Declaration("aggregate", AggregateNodeOptions(
        [("amount", "hash_sum", None, "revenue"), ("amount", "hash_count", None, "orders")],
        keys=["country"])),
    Declaration("order_by", OrderByNodeOptions([("revenue", "descending")])),
])

print(plan.to_table().to_pylist())
# [{'country': 'DE', 'revenue': 130.0, 'orders': 3}, {'country': 'US', 'revenue': 40.0, 'orders': 1}]

# pyarrow's Table.join and Table.group_by build the same kind of plan for you
print(orders.join(customers, "customer_id").group_by("country")
      .aggregate([("amount", "sum")]).sort_by("country").to_pylist())
# [{'country': 'DE', 'amount_sum': 130.0}, {'country': 'US', 'amount_sum': 45.0}]