Rust CLI Sales Analytics with PostgreSQL
Build a Rust CLI tool that runs advanced SQL analytics on a local PostgreSQL database.
Introduction
30 Second Summary
Every time you see a "trending products" list or a "your spending this month" breakdown, there's a database running analytical queries behind the scenes. These queries slice, rank, and compare data to surface the patterns that drive decisions.
In this project, you will build a command-line sales analytics tool in Rust that connects to a local PostgreSQL database and runs progressively advanced analytical queries against e-commerce data.
What You'll Build
A Rust CLI binary that creates an e-commerce schema in PostgreSQL, seeds it with sample data, and executes analytical queries that produce formatted reports in your terminal.
By the end of this project, you'll have:
- A Dockerized PostgreSQL instance with a multi-table e-commerce schema seeded with sample data.
- Aggregation and sorting queries using GROUP BY, multi-column ORDER BY, and NULLS FIRST/LAST handling.
- Window function and CTE queries that bucket customers into spending tiers and calculate moving averages with frame specifications.
- Secret Mission: Add month-over-month revenue growth analysis using LAG and a multi-dimensional ROLLUP report with category and monthly subtotals.
Are there any prerequisites?
This project assumes basic SQL knowledge (SELECT, INSERT, WHERE), basic Rust familiarity (variables, functions, structs), and comfort with the command line.
Set Up Docker and PostgreSQL
Every analytics tool needs data to analyze. Your Rust CLI will query a PostgreSQL database full of e-commerce sales data.
Docker makes this simple. Instead of installing PostgreSQL directly on your machine, you'll spin up a lightweight container that runs PostgreSQL 17. The database will be ready in seconds.
In this step, get ready to:
- Start a PostgreSQL 17 container with Docker.
- Verify the database accepts connections.
Start the PostgreSQL container
Docker containers run isolated services without polluting your system. You'll start a PostgreSQL 17 container that holds all the sales data your analytics tool will query.
- Start a PostgreSQL 17 container by running this command:
docker run --name pg-analytics -e POSTGRES_PASSWORD=analytics -e POSTGRES_DB=sales -e POSTGRES_USER=analyst -p 5432:5432 -d postgres:17-alpineWhat does this command do?
- --name pg-analytics gives the container a memorable name you can reference later.
- -e POSTGRES_PASSWORD=analytics sets the database password. This flag is required for the PostgreSQL Docker image.
- -e POSTGRES_USER=analyst creates a database user named analyst (defaults to postgres if not set).
- -e POSTGRES_DB=sales creates a database called sales at startup.
- -p 5432:5432 maps the container's PostgreSQL port to your machine so local programs can connect.
- -d postgres:17-alpine runs the container in the background using the lightweight Alpine variant of PostgreSQL 17.
- Confirm the container is running by running:
docker psYou should see a row with postgres:17-alpine in the IMAGE column and pg-analytics in the NAMES column. The STATUS column should show Up followed by how long the container has been running.
Don't see the container listed?
- If you see an error about the name pg-analytics already being in use, a previous container with that name exists. Remove it first with docker rm -f pg-analytics and then rerun the docker run command.
- If you see an error about port 5432 already being allocated, another service is using that port. Stop any other PostgreSQL instances or containers first.
Having trouble with Docker?
Test the database connection
A running container does not guarantee the database inside is ready. You'll connect directly and run a query to confirm PostgreSQL is accepting connections.
- Test the connection by running this command:
docker exec -it pg-analytics psql -U analyst -d sales -c "SELECT version();"What does this command do?
- docker exec -it runs a command inside a running container with an interactive terminal.
- psql -U analyst -d sales launches the PostgreSQL client as user analyst connected to the sales database.
- -c "SELECT version();" executes a single SQL query and exits. This query returns the PostgreSQL version string.
You should see output containing PostgreSQL 17 followed by version details. This confirms your database is running and accepting queries.
Seeing a connection refused error?
- PostgreSQL takes a few seconds to initialize after the container starts. Wait 5 seconds and try the command again.
- Make sure the container is still running with docker ps. If the STATUS shows Exited, check the logs with docker logs pg-analytics to see what went wrong.
Still stuck?
Your database is live and ready for data. Next up, you'll create the Rust project that connects to it.
Create the Rust Project and Connect to PostgreSQL
Your PostgreSQL database is running in Docker, waiting for queries. Now you need a Rust program that can reach it.
In this step, you'll scaffold a Rust project with Cargo, add the SQLx database library as a dependency, and write just enough code to confirm your program can connect to the database.
In this step, get ready to:
- Scaffold a new Rust project using Cargo.
- Configure SQLx and Tokio as dependencies.
- Verify your Rust program can connect to the PostgreSQL database.
Scaffold a new Rust project
Cargo is Rust's build tool and package manager. It can scaffold a project with the correct folder structure in one command.
- Move to your Desktop by running this command:
cd ~/Desktop- Scaffold the Rust project by running:
cargo init sales-analytics- Move into the project folder by running:
cd sales-analytics- Confirm the project files exist by running:
lsYou should see Cargo.toml and a src folder in the output.
- Open this folder in VS Code by clicking File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the sales-analytics folder, and click Open.
Don't see Cargo.toml in the output?
Make sure you ran cargo init sales-analytics from your Desktop directory. If you see an error about the directory already existing, delete it first with rm -rf sales-analytics and try again.
Having trouble?
Add SQLx and Tokio dependencies
SQLx is a Rust library that lets you run raw SQL queries against a database. Tokio is an async runtime that SQLx needs to handle database connections without blocking your program.
- In the VS Code Explorer sidebar, click Cargo.toml to open it.
- Replace everything in Cargo.toml with this configuration:
[package]
name = "sales-analytics"
version = "0.1.0"
edition = "2021"
[dependencies]
sqlx = { version = "0.9.0", features = ["runtime-tokio", "postgres", "bigdecimal", "chrono"] }
tokio = { version = "1.52.3", features = ["full"] }What do these dependencies do?
- sqlx with runtime-tokio and postgres features gives you async PostgreSQL support.
- The bigdecimal feature maps PostgreSQL NUMERIC columns to Rust types for precise money calculations.
- The chrono feature maps date and timestamp columns to Rust date types.
- tokio with full features provides the async runtime that SQLx uses under the hood.
- Save Cargo.toml by pressing Ctrl+S.
You won't see a visible change yet. The dependencies get downloaded when you first run the project in the next substep.
✔️ Awesome, I've got everything!
Great. Double check you've saved Cargo.toml and move on.
ⓧ I'd like to double check the full code
[package]
name = "sales-analytics"
version = "0.1.0"
edition = "2021"
[dependencies]
sqlx = { version = "0.9.0", features = ["runtime-tokio", "postgres", "bigdecimal", "chrono"] }
tokio = { version = "1.52.3", features = ["full"] }Write the connection code and verify it works
A connection pool manages multiple reusable database connections so your program does not open a new one every time it runs a query. SQLx provides PgPoolOptions to configure and create this pool.
- In the VS Code Explorer sidebar, expand the src folder and click main.rs to open it.
- Replace everything in src/main.rs with this code:
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
let database_url = "postgresql://analyst:analytics@localhost:5432/sales";
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(database_url)
.await?;
println!("Connected to PostgreSQL!");
Ok(())
}What does this code do?
- #[tokio::main] turns your main function into an async entry point so you can use .await for database calls.
- PgPoolOptions::new().max_connections(5) creates a pool that holds up to 5 reusable connections to your PostgreSQL database.
- The .connect(database_url).await? attempts the connection. If it fails (wrong password, database not running), the ? returns the error immediately.
- use sqlx::PgPool is imported for later steps where you will pass the pool to other functions.
- Save src/main.rs by pressing Ctrl+S.
- Run the program by opening a terminal in VS Code (Ctrl+`) and running:
cargo runFirst compile takes a while
Cargo downloads and compiles all dependencies on the first run. This can take 1-3 minutes depending on your machine. Subsequent runs will be much faster because Cargo caches the compiled dependencies.
After compilation finishes, you should see this output:
Connected to PostgreSQL!Seeing a connection error?
If you see an error mentioning a refused connection, your PostgreSQL container might not be running.
- Check if the container is running with docker ps. If you don't see pg-analytics in the list, start it again with docker start pg-analytics.
- If you see an authentication error, double check the database_url string in main.rs matches the user, password, and database name from Step 1.
Help me fix my PostgreSQL connection error in Rust.
✔️ Awesome, I've got everything!
Great. Double check you've saved src/main.rs and move on.
ⓧ I'd like to double check the full code
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
let database_url = "postgresql://analyst:analytics@localhost:5432/sales";
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(database_url)
.await?;
println!("Connected to PostgreSQL!");
Ok(())
}Your Rust project is connected to the database. Next up, you'll create the e-commerce schema and fill it with sample data so you have something to analyze.
Create the Schema and Seed Data
Your Rust program can connect to PostgreSQL, but the database is completely empty.
Analytics queries need data to analyze. In this step, you will design a multi-table e-commerce schema and populate it with sample sales data that spans multiple months.
In this step, get ready to:
- Write SQL to create four related tables with foreign keys.
- Seed the database with realistic e-commerce data.
- Run the program and confirm all data loaded successfully.
Write the schema creation function
A database schema defines the structure of your data. You will create four tables that model a simple e-commerce system: product categories, customers, orders, and individual order line items.
- In src/main.rs, find this import line near the top:
use sqlx::PgPool;- Replace it with this line to also import FromRow (you will use it in later steps):
use sqlx::{FromRow, PgPool};- Next, find the Ok(()) line inside your main function. Add these two function calls just above it:
create_schema(&pool).await?;
seed_data(&pool).await?;What do these calls do?
Each call passes a reference to the connection pool (&pool) so the functions can run SQL against the database. The await? pattern waits for the async operation to finish and propagates any errors upward.
- Now add the create_schema function below the closing brace of main. Copy the following code:
async fn create_schema(pool: &PgPool) -> Result<(), sqlx::Error> {
// Drop existing tables so we start fresh each run
sqlx::query("DROP TABLE IF EXISTS order_items, orders, customers, categories CASCADE")
.execute(pool)
.await?;
// Product categories (Electronics, Books, etc.)
sqlx::query(
"CREATE TABLE categories (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL
)"
)
.execute(pool)
.await?;
// Customer information
sqlx::query(
"CREATE TABLE customers (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
)"
)
.execute(pool)
.await?;
// Orders linked to customers via foreign key
sqlx::query(
"CREATE TABLE orders (
id SERIAL PRIMARY KEY,
customer_id INTEGER NOT NULL REFERENCES customers(id),
order_date TIMESTAMPTZ NOT NULL,
status VARCHAR(20) DEFAULT 'completed'
)"
)
.execute(pool)
.await?;
// Individual line items linking orders to categories
sqlx::query(
"CREATE TABLE order_items (
id SERIAL PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES orders(id),
category_id INTEGER NOT NULL REFERENCES categories(id),
product_name VARCHAR(200) NOT NULL,
quantity INTEGER NOT NULL,
unit_price NUMERIC(10,2) NOT NULL
)"
)
.execute(pool)
.await?;
println!("Schema created successfully!");
Ok(())
}What does this code do?
The function first drops any existing tables so you get a clean slate every time you run the program. Then it creates four tables in dependency order:
- categories and customers are independent tables with no foreign keys.
- orders references customers so every order belongs to a customer.
- order_items references both orders and categories, connecting line items to their order and product category.
SERIAL auto-generates incrementing IDs. NUMERIC(10,2) stores exact monetary values with two decimal places. TIMESTAMPTZ stores timestamps with timezone awareness.
You won't see a change yet. The next chunk adds the data these tables will hold.
Write the seed data function
Empty tables are no good for analytics. You need realistic data spanning different customers, categories, and time periods so your queries in later steps have meaningful patterns to uncover.
- Add the seed_data function below create_schema. Copy the following code:
async fn seed_data(pool: &PgPool) -> Result<(), sqlx::Error> {
// 4 product categories
sqlx::query(
"INSERT INTO categories (name) VALUES
('Electronics'), ('Books'), ('Clothing'), ('Home & Garden')"
)
.execute(pool)
.await?;
// 8 customers
sqlx::query(
"INSERT INTO customers (name, email) VALUES
('Alice Johnson', 'alice@example.com'),
('Bob Smith', 'bob@example.com'),
('Carol White', 'carol@example.com'),
('David Brown', 'david@example.com'),
('Eve Davis', 'eve@example.com'),
('Frank Miller', 'frank@example.com'),
('Grace Wilson', 'grace@example.com'),
('Henry Taylor', 'henry@example.com')"
)
.execute(pool)
.await?;
// 19 orders spread across Jan-Mar 2024
sqlx::query(
"INSERT INTO orders (customer_id, order_date) VALUES
(1, '2024-01-15'), (1, '2024-02-20'), (1, '2024-03-10'),
(2, '2024-01-22'), (2, '2024-03-05'),
(3, '2024-01-08'), (3, '2024-02-14'), (3, '2024-03-22'),
(4, '2024-02-01'), (4, '2024-03-15'),
(5, '2024-01-30'), (5, '2024-02-28'),
(6, '2024-01-12'), (6, '2024-02-18'), (6, '2024-03-25'),
(7, '2024-02-05'), (7, '2024-03-08'),
(8, '2024-01-25'), (8, '2024-03-20')"
)
.execute(pool)
.await?;
// 37 order items with varied products and prices
sqlx::query(
"INSERT INTO order_items (order_id, category_id, product_name, quantity, unit_price) VALUES
(1, 1, 'Laptop', 1, 999.99), (1, 1, 'Mouse', 2, 29.99),
(2, 2, 'Rust Programming', 1, 49.99), (2, 3, 'T-Shirt', 3, 19.99),
(3, 1, 'Keyboard', 1, 149.99), (3, 4, 'Desk Lamp', 2, 34.99),
(4, 1, 'Monitor', 1, 499.99), (4, 2, 'Database Internals', 1, 59.99),
(5, 3, 'Jacket', 1, 89.99), (5, 4, 'Plant Pot', 3, 15.99),
(6, 2, 'Clean Code', 2, 39.99), (6, 1, 'Webcam', 1, 79.99),
(7, 3, 'Dress', 2, 59.99), (7, 4, 'Garden Tools', 1, 45.99),
(8, 1, 'Tablet', 1, 349.99), (8, 3, 'Sneakers', 1, 129.99),
(9, 4, 'Bird Feeder', 2, 24.99), (9, 2, 'Design Patterns', 1, 54.99),
(10, 1, 'Headphones', 1, 199.99), (10, 3, 'Hat', 2, 14.99),
(11, 2, 'Algorithms', 1, 44.99), (11, 1, 'USB Hub', 1, 39.99),
(12, 4, 'Candle Set', 3, 19.99), (12, 3, 'Scarf', 1, 24.99),
(13, 1, 'Charger', 2, 24.99), (13, 2, 'Python Cookbook', 1, 34.99),
(14, 3, 'Jeans', 1, 69.99), (14, 4, 'Wall Clock', 1, 39.99),
(15, 1, 'SSD Drive', 1, 119.99), (15, 2, 'Go in Action', 1, 44.99),
(16, 3, 'Boots', 1, 149.99), (16, 4, 'Throw Pillow', 2, 29.99),
(17, 1, 'Power Bank', 2, 49.99), (17, 3, 'Socks Pack', 3, 12.99),
(18, 2, 'SQL Mastery', 1, 39.99), (18, 4, 'Vase', 1, 34.99),
(19, 1, 'Earbuds', 1, 79.99), (19, 3, 'Polo Shirt', 2, 34.99)"
)
.execute(pool)
.await?;
println!("Seeded: 4 categories, 8 customers, 19 orders, 37 order items");
Ok(())
}What does this code do?
The function inserts sample data in the same dependency order as the schema. Categories and customers go in first because orders and order_items reference them via foreign keys.
The data spans January through March 2024, with 8 customers placing 19 orders containing 37 line items. This gives you enough volume and variety for meaningful aggregation, sorting, and window function queries in the next steps.
Run the program and verify the data
- Save src/main.rs.
- Run the program by executing this command in the terminal (make sure you are inside the sales-analytics folder):
cargo runYou should see three lines of output confirming connection, schema creation, and data seeding:
Connected to PostgreSQL!
Schema created successfully!
Seeded: 4 categories, 8 customers, 19 orders, 37 order itemsSeeing a connection error?
- Make sure your Docker container is still running. Run docker ps and check for pg-analytics in the list. If it stopped, start it again with docker start pg-analytics.
- Check that your connection string matches exactly: postgresql://analyst:analytics@localhost:5432/sales.
- If you see a compile error about create_schema or seed_data not found, make sure the functions are defined outside of main (below its closing brace).
Still stuck? Help me debug my schema creation and seeding errors.
✔️ Awesome, I've got everything!
Great. Double check you have saved src/main.rs and your terminal shows all three success messages.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs at the end of this step:
use sqlx::postgres::PgPoolOptions;
use sqlx::{FromRow, PgPool};
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
let database_url = "postgresql://analyst:analytics@localhost:5432/sales";
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(database_url)
.await?;
println!("Connected to PostgreSQL!");
create_schema(&pool).await?;
seed_data(&pool).await?;
Ok(())
}
async fn create_schema(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query("DROP TABLE IF EXISTS order_items, orders, customers, categories CASCADE")
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE categories (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE customers (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE orders (
id SERIAL PRIMARY KEY,
customer_id INTEGER NOT NULL REFERENCES customers(id),
order_date TIMESTAMPTZ NOT NULL,
status VARCHAR(20) DEFAULT 'completed'
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE order_items (
id SERIAL PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES orders(id),
category_id INTEGER NOT NULL REFERENCES categories(id),
product_name VARCHAR(200) NOT NULL,
quantity INTEGER NOT NULL,
unit_price NUMERIC(10,2) NOT NULL
)"
)
.execute(pool)
.await?;
println!("Schema created successfully!");
Ok(())
}
async fn seed_data(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO categories (name) VALUES
('Electronics'), ('Books'), ('Clothing'), ('Home & Garden')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO customers (name, email) VALUES
('Alice Johnson', 'alice@example.com'),
('Bob Smith', 'bob@example.com'),
('Carol White', 'carol@example.com'),
('David Brown', 'david@example.com'),
('Eve Davis', 'eve@example.com'),
('Frank Miller', 'frank@example.com'),
('Grace Wilson', 'grace@example.com'),
('Henry Taylor', 'henry@example.com')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO orders (customer_id, order_date) VALUES
(1, '2024-01-15'), (1, '2024-02-20'), (1, '2024-03-10'),
(2, '2024-01-22'), (2, '2024-03-05'),
(3, '2024-01-08'), (3, '2024-02-14'), (3, '2024-03-22'),
(4, '2024-02-01'), (4, '2024-03-15'),
(5, '2024-01-30'), (5, '2024-02-28'),
(6, '2024-01-12'), (6, '2024-02-18'), (6, '2024-03-25'),
(7, '2024-02-05'), (7, '2024-03-08'),
(8, '2024-01-25'), (8, '2024-03-20')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO order_items (order_id, category_id, product_name, quantity, unit_price) VALUES
(1, 1, 'Laptop', 1, 999.99), (1, 1, 'Mouse', 2, 29.99),
(2, 2, 'Rust Programming', 1, 49.99), (2, 3, 'T-Shirt', 3, 19.99),
(3, 1, 'Keyboard', 1, 149.99), (3, 4, 'Desk Lamp', 2, 34.99),
(4, 1, 'Monitor', 1, 499.99), (4, 2, 'Database Internals', 1, 59.99),
(5, 3, 'Jacket', 1, 89.99), (5, 4, 'Plant Pot', 3, 15.99),
(6, 2, 'Clean Code', 2, 39.99), (6, 1, 'Webcam', 1, 79.99),
(7, 3, 'Dress', 2, 59.99), (7, 4, 'Garden Tools', 1, 45.99),
(8, 1, 'Tablet', 1, 349.99), (8, 3, 'Sneakers', 1, 129.99),
(9, 4, 'Bird Feeder', 2, 24.99), (9, 2, 'Design Patterns', 1, 54.99),
(10, 1, 'Headphones', 1, 199.99), (10, 3, 'Hat', 2, 14.99),
(11, 2, 'Algorithms', 1, 44.99), (11, 1, 'USB Hub', 1, 39.99),
(12, 4, 'Candle Set', 3, 19.99), (12, 3, 'Scarf', 1, 24.99),
(13, 1, 'Charger', 2, 24.99), (13, 2, 'Python Cookbook', 1, 34.99),
(14, 3, 'Jeans', 1, 69.99), (14, 4, 'Wall Clock', 1, 39.99),
(15, 1, 'SSD Drive', 1, 119.99), (15, 2, 'Go in Action', 1, 44.99),
(16, 3, 'Boots', 1, 149.99), (16, 4, 'Throw Pillow', 2, 29.99),
(17, 1, 'Power Bank', 2, 49.99), (17, 3, 'Socks Pack', 3, 12.99),
(18, 2, 'SQL Mastery', 1, 39.99), (18, 4, 'Vase', 1, 34.99),
(19, 1, 'Earbuds', 1, 79.99), (19, 3, 'Polo Shirt', 2, 34.99)"
)
.execute(pool)
.await?;
println!("Seeded: 4 categories, 8 customers, 19 orders, 37 order items");
Ok(())
}Your database now has a complete e-commerce dataset ready for analysis. Next up, you will write your first analytical queries to surface revenue patterns and customer rankings from this data.
Write Aggregation and Sorting Queries
Your database is loaded with e-commerce data across four tables. Now it's time to ask questions of that data.
In this step, you'll write SQL aggregation queries that JOIN tables, GROUP results, and sort them with multi-column ORDER BY. You'll also learn how PostgreSQL handles NULL values in sorting, which matters whenever real-world data has gaps.
In this step, get ready to:
- Define Rust structs that map SQL query results into typed data.
- Write GROUP BY aggregation queries with multi-column sorting.
- Demonstrate NULLS LAST sorting to handle customers with missing order dates.
Define result structs for query mapping
SQLx maps each row of query results into a Rust struct using the FromRow derive macro. Each struct field name must match the column alias in your SQL query.
- In src/main.rs, find the blank line between the use statements and #[tokio::main].
- Add these three struct definitions between them:
#[derive(FromRow)]
struct RevenueByCat {
category_name: String,
total_revenue: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct TopCustomer {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct CustomerLastOrder {
customer_name: String,
last_order_date: Option<chrono::NaiveDate>,
total_spend: sqlx::types::BigDecimal,
}What do these structs do?
Each struct represents one row of a specific query's results. SQLx automatically maps column values to struct fields by matching names.
- sqlx::types::BigDecimal maps to PostgreSQL's NUMERIC type, preserving decimal precision for money calculations.
- Option<chrono::NaiveDate> handles cases where a date column might be NULL (e.g., a customer who has never ordered).
- The i64 type maps to PostgreSQL's COUNT aggregate, which returns a bigint.
- Save src/main.rs.
- You won't see a change yet. The next chunk uses these structs in actual queries.
Write the Revenue by Category query
Your first analytics query answers: which product categories generate the most revenue? This combines JOIN, GROUP BY, SUM, and multi-column ORDER BY.
- At the bottom of src/main.rs (after the closing } of seed_data), add this new function:
async fn run_aggregation_queries(pool: &PgPool) -> Result<(), sqlx::Error> {
println!("\n============================================================");
println!(" SALES ANALYTICS REPORT");
println!("============================================================");
// Revenue by Category with multi-column sort
println!("\n--- Revenue by Category (multi-column sort) ---\n");
let results = sqlx::query_as::<_, RevenueByCat>(
"SELECT
c.name AS category_name,
SUM(oi.quantity * oi.unit_price) AS total_revenue,
COUNT(DISTINCT o.id) AS order_count
FROM order_items oi
JOIN categories c ON c.id = oi.category_id
JOIN orders o ON o.id = oi.order_id
GROUP BY c.name
ORDER BY total_revenue DESC, category_name ASC"
)
.fetch_all(pool)
.await?;
println!("{:<15} {:>15} {:>12}", "Category", "Revenue", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<15} {:>15} {:>12}", row.category_name, row.total_revenue, row.order_count);
}
Ok(())
}What does this query do?
- JOIN connects three tables: order_items (where prices live), categories (for names), and orders (to count distinct orders).
- GROUP BY c.name collapses all rows for each category into one summary row.
- ORDER BY total_revenue DESC, category_name ASC sorts highest revenue first. If two categories tie on revenue, they sort alphabetically.
- sqlx::query_as::<_, RevenueByCat> tells SQLx to map each result row into your RevenueByCat struct.
- Save src/main.rs.
- You won't run this yet. The next chunk adds the second query before you wire everything up.
Now add a second query that finds the highest-spending customers.
- In the run_aggregation_queries function, find the line that says Ok(()).
- Add this code above Ok(()):
// Top 5 Customers
println!("\n--- Top 5 Customers by Spend ---\n");
let results = sqlx::query_as::<_, TopCustomer>(
"SELECT
cu.name AS customer_name,
SUM(oi.quantity * oi.unit_price) AS total_spend,
COUNT(DISTINCT o.id) AS order_count
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY total_spend DESC
LIMIT 5"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>10}", "Customer", "Spend", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<20} {:>12} {:>10}", row.customer_name, row.total_spend, row.order_count);
}What does this query do?
- This query JOINs customers to their orders and order items, then groups by customer name to sum their total spending.
- LIMIT 5 caps the results at the top five spenders after sorting by total_spend DESC.
- COUNT(DISTINCT o.id) counts unique orders, not individual line items. Without DISTINCT, an order with three items would count as three.
- Save src/main.rs.
- You won't see a change yet. The next substep adds the final query and connects everything to main.
Add NULLS LAST sorting and run the report
Real data has gaps. A customer might exist in your database but have no orders yet. PostgreSQL's NULLS LAST clause lets you control where those NULL values land in sorted results.
- In run_aggregation_queries, find Ok(()) again and add this code above it:
// NULLS LAST demonstration
println!("\n--- Customer Activity (NULLS LAST sorting) ---\n");
let results = sqlx::query_as::<_, CustomerLastOrder>(
"SELECT
cu.name AS customer_name,
MAX(o.order_date)::DATE AS last_order_date,
COALESCE(SUM(oi.quantity * oi.unit_price), 0) AS total_spend
FROM customers cu
LEFT JOIN orders o ON o.customer_id = cu.id
LEFT JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY last_order_date DESC NULLS LAST, total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>14} {:>12}", "Customer", "Last Order", "Total Spend");
println!("{}", "-".repeat(48));
for row in &results {
let date_str = match row.last_order_date {
Some(d) => d.to_string(),
None => "Never".to_string(),
};
println!("{:<20} {:>14} {:>12}", row.customer_name, date_str, row.total_spend);
}What does this query do?
- LEFT JOIN keeps all customers in the results, even those with zero orders. A regular JOIN would silently drop them.
- MAX(o.order_date)::DATE finds each customer's most recent order. If they have no orders, this returns NULL.
- COALESCE(SUM(...), 0) replaces a NULL sum with 0 for customers who never ordered.
- ORDER BY last_order_date DESC NULLS LAST sorts the most recent activity first, pushing customers with no orders (NULL dates) to the bottom of the list.
Why NULLS LAST matters
Without NULLS LAST, PostgreSQL places NULLs at the top when sorting DESC. That means inactive customers would appear before your most active ones. NULLS LAST overrides this default so that missing data sinks to the bottom where it belongs in a "most recent activity" report.
Now wire the new function into your program's main flow.
- In the main function, find this line:
seed_data(&pool).await?;- Add this line directly below it:
run_aggregation_queries(&pool).await?;- Save src/main.rs.
- Run the program by running this command in your terminal:
cargo run- You should see the "SALES ANALYTICS REPORT" header followed by three formatted tables: Revenue by Category, Top 5 Customers by Spend, and Customer Activity with NULLS LAST sorting.
Seeing a compile error?
- If you see "cannot find type" errors for RevenueByCat or the other structs, make sure the struct definitions are above the main function (between the use statements and #[tokio::main]).
- If you see "mismatched types" on a struct field, check that your SQL column aliases (the AS names) match your struct field names exactly.
- If the query runs but shows zero rows for Revenue by Category, make sure the seed_data call still appears before run_aggregation_queries in main.
Help me debug my cargo run output for the aggregation queries.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here's the complete src/main.rs at this point:
use sqlx::postgres::PgPoolOptions;
use sqlx::{FromRow, PgPool};
#[derive(FromRow)]
struct RevenueByCat {
category_name: String,
total_revenue: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct TopCustomer {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct CustomerLastOrder {
customer_name: String,
last_order_date: Option<chrono::NaiveDate>,
total_spend: sqlx::types::BigDecimal,
}
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
let database_url = "postgresql://analyst:analytics@localhost:5432/sales";
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(database_url)
.await?;
println!("Connected to PostgreSQL!");
create_schema(&pool).await?;
seed_data(&pool).await?;
run_aggregation_queries(&pool).await?;
Ok(())
}
async fn create_schema(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query("DROP TABLE IF EXISTS order_items, orders, customers, categories CASCADE")
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE categories (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE customers (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE orders (
id SERIAL PRIMARY KEY,
customer_id INTEGER NOT NULL REFERENCES customers(id),
order_date TIMESTAMPTZ NOT NULL,
status VARCHAR(20) DEFAULT 'completed'
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE order_items (
id SERIAL PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES orders(id),
category_id INTEGER NOT NULL REFERENCES categories(id),
product_name VARCHAR(200) NOT NULL,
quantity INTEGER NOT NULL,
unit_price NUMERIC(10,2) NOT NULL
)"
)
.execute(pool)
.await?;
println!("Schema created successfully!");
Ok(())
}
async fn seed_data(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO categories (name) VALUES
('Electronics'), ('Books'), ('Clothing'), ('Home & Garden')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO customers (name, email) VALUES
('Alice Johnson', 'alice@example.com'),
('Bob Smith', 'bob@example.com'),
('Carol White', 'carol@example.com'),
('David Brown', 'david@example.com'),
('Eve Davis', 'eve@example.com'),
('Frank Miller', 'frank@example.com'),
('Grace Wilson', 'grace@example.com'),
('Henry Taylor', 'henry@example.com')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO orders (customer_id, order_date) VALUES
(1, '2024-01-15'), (1, '2024-02-20'), (1, '2024-03-10'),
(2, '2024-01-22'), (2, '2024-03-05'),
(3, '2024-01-08'), (3, '2024-02-14'), (3, '2024-03-22'),
(4, '2024-02-01'), (4, '2024-03-15'),
(5, '2024-01-30'), (5, '2024-02-28'),
(6, '2024-01-12'), (6, '2024-02-18'), (6, '2024-03-25'),
(7, '2024-02-05'), (7, '2024-03-08'),
(8, '2024-01-25'), (8, '2024-03-20')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO order_items (order_id, category_id, product_name, quantity, unit_price) VALUES
(1, 1, 'Laptop', 1, 999.99), (1, 1, 'Mouse', 2, 29.99),
(2, 2, 'Rust Programming', 1, 49.99), (2, 3, 'T-Shirt', 3, 19.99),
(3, 1, 'Keyboard', 1, 149.99), (3, 4, 'Desk Lamp', 2, 34.99),
(4, 1, 'Monitor', 1, 499.99), (4, 2, 'Database Internals', 1, 59.99),
(5, 3, 'Jacket', 1, 89.99), (5, 4, 'Plant Pot', 3, 15.99),
(6, 2, 'Clean Code', 2, 39.99), (6, 1, 'Webcam', 1, 79.99),
(7, 3, 'Dress', 2, 59.99), (7, 4, 'Garden Tools', 1, 45.99),
(8, 1, 'Tablet', 1, 349.99), (8, 3, 'Sneakers', 1, 129.99),
(9, 4, 'Bird Feeder', 2, 24.99), (9, 2, 'Design Patterns', 1, 54.99),
(10, 1, 'Headphones', 1, 199.99), (10, 3, 'Hat', 2, 14.99),
(11, 2, 'Algorithms', 1, 44.99), (11, 1, 'USB Hub', 1, 39.99),
(12, 4, 'Candle Set', 3, 19.99), (12, 3, 'Scarf', 1, 24.99),
(13, 1, 'Charger', 2, 24.99), (13, 2, 'Python Cookbook', 1, 34.99),
(14, 3, 'Jeans', 1, 69.99), (14, 4, 'Wall Clock', 1, 39.99),
(15, 1, 'SSD Drive', 1, 119.99), (15, 2, 'Go in Action', 1, 44.99),
(16, 3, 'Boots', 1, 149.99), (16, 4, 'Throw Pillow', 2, 29.99),
(17, 1, 'Power Bank', 2, 49.99), (17, 3, 'Socks Pack', 3, 12.99),
(18, 2, 'SQL Mastery', 1, 39.99), (18, 4, 'Vase', 1, 34.99),
(19, 1, 'Earbuds', 1, 79.99), (19, 3, 'Polo Shirt', 2, 34.99)"
)
.execute(pool)
.await?;
println!("Seeded: 4 categories, 8 customers, 19 orders, 37 order items");
Ok(())
}
async fn run_aggregation_queries(pool: &PgPool) -> Result<(), sqlx::Error> {
println!("\n============================================================");
println!(" SALES ANALYTICS REPORT");
println!("============================================================");
// Revenue by Category with multi-column sort
println!("\n--- Revenue by Category (multi-column sort) ---\n");
let results = sqlx::query_as::<_, RevenueByCat>(
"SELECT
c.name AS category_name,
SUM(oi.quantity * oi.unit_price) AS total_revenue,
COUNT(DISTINCT o.id) AS order_count
FROM order_items oi
JOIN categories c ON c.id = oi.category_id
JOIN orders o ON o.id = oi.order_id
GROUP BY c.name
ORDER BY total_revenue DESC, category_name ASC"
)
.fetch_all(pool)
.await?;
println!("{:<15} {:>15} {:>12}", "Category", "Revenue", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<15} {:>15} {:>12}", row.category_name, row.total_revenue, row.order_count);
}
// Top 5 Customers
println!("\n--- Top 5 Customers by Spend ---\n");
let results = sqlx::query_as::<_, TopCustomer>(
"SELECT
cu.name AS customer_name,
SUM(oi.quantity * oi.unit_price) AS total_spend,
COUNT(DISTINCT o.id) AS order_count
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY total_spend DESC
LIMIT 5"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>10}", "Customer", "Spend", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<20} {:>12} {:>10}", row.customer_name, row.total_spend, row.order_count);
}
// NULLS LAST demonstration
println!("\n--- Customer Activity (NULLS LAST sorting) ---\n");
let results = sqlx::query_as::<_, CustomerLastOrder>(
"SELECT
cu.name AS customer_name,
MAX(o.order_date)::DATE AS last_order_date,
COALESCE(SUM(oi.quantity * oi.unit_price), 0) AS total_spend
FROM customers cu
LEFT JOIN orders o ON o.customer_id = cu.id
LEFT JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY last_order_date DESC NULLS LAST, total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>14} {:>12}", "Customer", "Last Order", "Total Spend");
println!("{}", "-".repeat(48));
for row in &results {
let date_str = match row.last_order_date {
Some(d) => d.to_string(),
None => "Never".to_string(),
};
println!("{:<20} {:>14} {:>12}", row.customer_name, date_str, row.total_spend);
}
Ok(())
}Your analytics tool now produces real reports from raw data. Next up, you'll write even more powerful queries using Common Table Expressions and window functions to unlock insights that flat GROUP BY can't reach.
Write CTE, Window Function, and Moving Average Queries
Your aggregation queries from the previous step answer "how much" questions. Now you need queries that answer "compared to what" questions.
In this step, you will write three progressively advanced queries using Common Table Expressions (CTEs), the NTILE window function for bucketing, and a moving average with an explicit window frame specification. These are the patterns data engineers use daily to segment users, calculate rolling metrics, and break complex logic into readable steps.
In this step, get ready to:
- Define result structs for advanced analytics queries.
- Write a CTE for customer lifetime value and an NTILE query for spending tiers.
- Write a 3-day moving average query using ROWS BETWEEN frame specification.
Define result structs for the advanced queries
Each new query returns a different shape of data. You need Rust structs that match the columns each query produces so SQLx can map the rows into typed values.
- In src/main.rs, find the CustomerLastOrder struct.
- Add these three new structs below it:
#[derive(FromRow)]
struct CustomerLifetime {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
order_count: i64,
avg_order_value: sqlx::types::BigDecimal,
}
#[derive(FromRow)]
struct SpendingTier {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
tier: i32,
}
#[derive(FromRow)]
struct MovingAvgRevenue {
order_date: chrono::NaiveDate,
daily_revenue: sqlx::types::BigDecimal,
moving_avg_3day: Option<sqlx::types::BigDecimal>,
}What do these structs represent?
- CustomerLifetime holds a customer's total spend, order count, and average order value. The CTE query produces these columns.
- SpendingTier holds a customer name, their spend, and which quartile bucket (1-4) they fall into.
- MovingAvgRevenue holds a date, that day's revenue, and the rolling 3-day average. The average is Option<BigDecimal> because the first row has no preceding rows to average.
- Save src/main.rs.
You won't see a change yet. The next chunk uses these structs in actual queries.
Write the CTE and NTILE queries
A CTE lets you name a subquery and reference it like a table. Instead of nesting subqueries inside subqueries, you write each logical step as its own named block.
- At the bottom of src/main.rs (after the closing } of run_aggregation_queries), add the beginning of a new run_advanced_queries function with the CTE query:
async fn run_advanced_queries(pool: &PgPool) -> Result<(), sqlx::Error> {
// CTE: Customer Lifetime Value
println!("\n--- Customer Lifetime Value (CTE) ---\n");
let results = sqlx::query_as::<_, CustomerLifetime>(
"WITH customer_orders AS (
SELECT
cu.name AS customer_name,
o.id AS order_id,
SUM(oi.quantity * oi.unit_price) AS order_total
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name, o.id
)
SELECT
customer_name,
SUM(order_total) AS total_spend,
COUNT(order_id) AS order_count,
ROUND(AVG(order_total), 2) AS avg_order_value
FROM customer_orders
GROUP BY customer_name
ORDER BY total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>8} {:>12}", "Customer", "Total", "Orders", "Avg Order");
println!("{}", "-".repeat(56));
for row in &results {
println!("{:<20} {:>12} {:>8} {:>12}", row.customer_name, row.total_spend, row.order_count, row.avg_order_value);
}What does this CTE do?
The WITH customer_orders AS (...) block creates a temporary named result set. It calculates the total for each individual order per customer.
The outer SELECT then aggregates those per-order totals into lifetime metrics: total spend, number of orders, and average order value. Breaking this into two steps makes the logic readable.
- Directly below the CTE query code (still inside the same function), add the NTILE spending tiers query:
// NTILE: Customer Spending Tiers
println!("\n--- Customer Spending Tiers (NTILE) ---\n");
let results = sqlx::query_as::<_, SpendingTier>(
"WITH customer_totals AS (
SELECT
cu.name AS customer_name,
SUM(oi.quantity * oi.unit_price) AS total_spend
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
)
SELECT
customer_name,
total_spend,
NTILE(4) OVER (ORDER BY total_spend DESC) AS tier
FROM customer_totals
ORDER BY tier, total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>10}", "Customer", "Spend", "Tier");
println!("{}", "-".repeat(44));
for row in &results {
let tier_label = match row.tier {
1 => "Top 25%",
2 => "High",
3 => "Medium",
4 => "Low",
_ => "Unknown",
};
println!("{:<20} {:>12} {:>10}", row.customer_name, row.total_spend, tier_label);
}What is NTILE and why use it here?
NTILE(4) is a window function that divides all rows into 4 equal-sized buckets based on the ORDER BY clause. With 8 customers divided into 4 tiers, each tier gets exactly 2 customers.
The CTE first calculates each customer's total spend. Then the outer query applies NTILE(4) OVER (ORDER BY total_spend DESC) to assign tier numbers (1 = top spenders, 4 = lowest). The Rust match converts those numbers into readable labels.
- Save src/main.rs.
You won't see output yet because the function isn't complete or called from main. The next chunk finishes it.
Write the moving average query and run the program
A moving average smooths out day-to-day spikes by averaging each row with its neighbors. The key is the frame specification, which tells PostgreSQL exactly which rows to include in each calculation.
- Still inside run_advanced_queries, add the moving average query below the NTILE code, then close the function with Ok(()):
// Moving Average with ROWS BETWEEN
println!("\n--- 3-Day Moving Average Revenue (ROWS BETWEEN) ---\n");
let results = sqlx::query_as::<_, MovingAvgRevenue>(
"WITH daily_rev AS (
SELECT
o.order_date::DATE AS order_date,
SUM(oi.quantity * oi.unit_price) AS daily_revenue
FROM orders o
JOIN order_items oi ON oi.order_id = o.id
GROUP BY o.order_date::DATE
)
SELECT
order_date,
daily_revenue,
ROUND(AVG(daily_revenue) OVER (
ORDER BY order_date
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
), 2) AS moving_avg_3day
FROM daily_rev
ORDER BY order_date"
)
.fetch_all(pool)
.await?;
println!("{:<12} {:>14} {:>16}", "Date", "Daily Rev", "3-Day Avg");
println!("{}", "-".repeat(44));
for row in &results {
let avg_str = match &row.moving_avg_3day {
Some(v) => format!("{}", v),
None => "N/A".to_string(),
};
println!("{:<12} {:>14} {:>16}", row.order_date, row.daily_revenue, avg_str);
}
Ok(())
}What does ROWS BETWEEN 2 PRECEDING AND CURRENT ROW do?
This is a window frame specification. It tells PostgreSQL to calculate the AVG using only the current row plus the 2 rows immediately before it (ordered by date). Each row gets a "sliding window" of at most 3 days.
Without an explicit frame, PostgreSQL defaults to RANGE UNBOUNDED PRECEDING, which would give you a cumulative average of all rows up to the current one. Specifying ROWS BETWEEN gives you precise control over the window size.
- Now connect the new function to main. Find this line in your main function:
run_aggregation_queries(&pool).await?;
Ok(())- Add the run_advanced_queries call between the aggregation call and Ok(()), so it looks like this:
run_aggregation_queries(&pool).await?;
run_advanced_queries(&pool).await?;
Ok(())- Save src/main.rs.
- Run the program to see all three new analytics sections by running:
cargo runYou should see three new sections in the output: a Customer Lifetime Value table, a Customer Spending Tiers table with labels like "Top 25%" and "Low", and a 3-Day Moving Average table showing daily revenue alongside the rolling average.
Seeing a compile error?
- If you see "cannot find type" errors, check that all three structs (CustomerLifetime, SpendingTier, MovingAvgRevenue) are placed before the main function, not inside it.
- If you see "mismatched types" on tier, make sure the field type is i32 (not i64). PostgreSQL NTILE returns a 32-bit integer.
- If the function compiles but returns no rows, verify that run_advanced_queries(&pool).await?; appears after the aggregation call in main.
Help me debug a compile error in my run_advanced_queries function
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs at this point:
use sqlx::postgres::PgPoolOptions;
use sqlx::{FromRow, PgPool};
#[derive(FromRow)]
struct RevenueByCat {
category_name: String,
total_revenue: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct TopCustomer {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
order_count: i64,
}
#[derive(FromRow)]
struct CustomerLastOrder {
customer_name: String,
last_order_date: Option<chrono::NaiveDate>,
total_spend: sqlx::types::BigDecimal,
}
#[derive(FromRow)]
struct CustomerLifetime {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
order_count: i64,
avg_order_value: sqlx::types::BigDecimal,
}
#[derive(FromRow)]
struct SpendingTier {
customer_name: String,
total_spend: sqlx::types::BigDecimal,
tier: i32,
}
#[derive(FromRow)]
struct MovingAvgRevenue {
order_date: chrono::NaiveDate,
daily_revenue: sqlx::types::BigDecimal,
moving_avg_3day: Option<sqlx::types::BigDecimal>,
}
#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
let database_url = "postgresql://analyst:analytics@localhost:5432/sales";
let pool = PgPoolOptions::new()
.max_connections(5)
.connect(database_url)
.await?;
println!("Connected to PostgreSQL!");
create_schema(&pool).await?;
seed_data(&pool).await?;
run_aggregation_queries(&pool).await?;
run_advanced_queries(&pool).await?;
Ok(())
}
async fn create_schema(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query("DROP TABLE IF EXISTS order_items, orders, customers, categories CASCADE")
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE categories (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE customers (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE orders (
id SERIAL PRIMARY KEY,
customer_id INTEGER NOT NULL REFERENCES customers(id),
order_date TIMESTAMPTZ NOT NULL,
status VARCHAR(20) DEFAULT 'completed'
)"
)
.execute(pool)
.await?;
sqlx::query(
"CREATE TABLE order_items (
id SERIAL PRIMARY KEY,
order_id INTEGER NOT NULL REFERENCES orders(id),
category_id INTEGER NOT NULL REFERENCES categories(id),
product_name VARCHAR(200) NOT NULL,
quantity INTEGER NOT NULL,
unit_price NUMERIC(10,2) NOT NULL
)"
)
.execute(pool)
.await?;
println!("Schema created successfully!");
Ok(())
}
async fn seed_data(pool: &PgPool) -> Result<(), sqlx::Error> {
sqlx::query(
"INSERT INTO categories (name) VALUES
('Electronics'), ('Books'), ('Clothing'), ('Home & Garden')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO customers (name, email) VALUES
('Alice Johnson', 'alice@example.com'),
('Bob Smith', 'bob@example.com'),
('Carol White', 'carol@example.com'),
('David Brown', 'david@example.com'),
('Eve Davis', 'eve@example.com'),
('Frank Miller', 'frank@example.com'),
('Grace Wilson', 'grace@example.com'),
('Henry Taylor', 'henry@example.com')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO orders (customer_id, order_date) VALUES
(1, '2024-01-15'), (1, '2024-02-20'), (1, '2024-03-10'),
(2, '2024-01-22'), (2, '2024-03-05'),
(3, '2024-01-08'), (3, '2024-02-14'), (3, '2024-03-22'),
(4, '2024-02-01'), (4, '2024-03-15'),
(5, '2024-01-30'), (5, '2024-02-28'),
(6, '2024-01-12'), (6, '2024-02-18'), (6, '2024-03-25'),
(7, '2024-02-05'), (7, '2024-03-08'),
(8, '2024-01-25'), (8, '2024-03-20')"
)
.execute(pool)
.await?;
sqlx::query(
"INSERT INTO order_items (order_id, category_id, product_name, quantity, unit_price) VALUES
(1, 1, 'Laptop', 1, 999.99), (1, 1, 'Mouse', 2, 29.99),
(2, 2, 'Rust Programming', 1, 49.99), (2, 3, 'T-Shirt', 3, 19.99),
(3, 1, 'Keyboard', 1, 149.99), (3, 4, 'Desk Lamp', 2, 34.99),
(4, 1, 'Monitor', 1, 499.99), (4, 2, 'Database Internals', 1, 59.99),
(5, 3, 'Jacket', 1, 89.99), (5, 4, 'Plant Pot', 3, 15.99),
(6, 2, 'Clean Code', 2, 39.99), (6, 1, 'Webcam', 1, 79.99),
(7, 3, 'Dress', 2, 59.99), (7, 4, 'Garden Tools', 1, 45.99),
(8, 1, 'Tablet', 1, 349.99), (8, 3, 'Sneakers', 1, 129.99),
(9, 4, 'Bird Feeder', 2, 24.99), (9, 2, 'Design Patterns', 1, 54.99),
(10, 1, 'Headphones', 1, 199.99), (10, 3, 'Hat', 2, 14.99),
(11, 2, 'Algorithms', 1, 44.99), (11, 1, 'USB Hub', 1, 39.99),
(12, 4, 'Candle Set', 3, 19.99), (12, 3, 'Scarf', 1, 24.99),
(13, 1, 'Charger', 2, 24.99), (13, 2, 'Python Cookbook', 1, 34.99),
(14, 3, 'Jeans', 1, 69.99), (14, 4, 'Wall Clock', 1, 39.99),
(15, 1, 'SSD Drive', 1, 119.99), (15, 2, 'Go in Action', 1, 44.99),
(16, 3, 'Boots', 1, 149.99), (16, 4, 'Throw Pillow', 2, 29.99),
(17, 1, 'Power Bank', 2, 49.99), (17, 3, 'Socks Pack', 3, 12.99),
(18, 2, 'SQL Mastery', 1, 39.99), (18, 4, 'Vase', 1, 34.99),
(19, 1, 'Earbuds', 1, 79.99), (19, 3, 'Polo Shirt', 2, 34.99)"
)
.execute(pool)
.await?;
println!("Seeded: 4 categories, 8 customers, 19 orders, 37 order items");
Ok(())
}
async fn run_aggregation_queries(pool: &PgPool) -> Result<(), sqlx::Error> {
println!("\n============================================================");
println!(" SALES ANALYTICS REPORT");
println!("============================================================");
// Revenue by Category with multi-column sort
println!("\n--- Revenue by Category (multi-column sort) ---\n");
let results = sqlx::query_as::<_, RevenueByCat>(
"SELECT
c.name AS category_name,
SUM(oi.quantity * oi.unit_price) AS total_revenue,
COUNT(DISTINCT o.id) AS order_count
FROM order_items oi
JOIN categories c ON c.id = oi.category_id
JOIN orders o ON o.id = oi.order_id
GROUP BY c.name
ORDER BY total_revenue DESC, category_name ASC"
)
.fetch_all(pool)
.await?;
println!("{:<15} {:>15} {:>12}", "Category", "Revenue", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<15} {:>15} {:>12}", row.category_name, row.total_revenue, row.order_count);
}
// Top 5 Customers
println!("\n--- Top 5 Customers by Spend ---\n");
let results = sqlx::query_as::<_, TopCustomer>(
"SELECT
cu.name AS customer_name,
SUM(oi.quantity * oi.unit_price) AS total_spend,
COUNT(DISTINCT o.id) AS order_count
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY total_spend DESC
LIMIT 5"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>10}", "Customer", "Spend", "Orders");
println!("{}", "-".repeat(44));
for row in &results {
println!("{:<20} {:>12} {:>10}", row.customer_name, row.total_spend, row.order_count);
}
// NULLS LAST demonstration
println!("\n--- Customer Activity (NULLS LAST sorting) ---\n");
let results = sqlx::query_as::<_, CustomerLastOrder>(
"SELECT
cu.name AS customer_name,
MAX(o.order_date)::DATE AS last_order_date,
COALESCE(SUM(oi.quantity * oi.unit_price), 0) AS total_spend
FROM customers cu
LEFT JOIN orders o ON o.customer_id = cu.id
LEFT JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
ORDER BY last_order_date DESC NULLS LAST, total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>14} {:>12}", "Customer", "Last Order", "Total Spend");
println!("{}", "-".repeat(48));
for row in &results {
let date_str = match row.last_order_date {
Some(d) => d.to_string(),
None => "Never".to_string(),
};
println!("{:<20} {:>14} {:>12}", row.customer_name, date_str, row.total_spend);
}
Ok(())
}
async fn run_advanced_queries(pool: &PgPool) -> Result<(), sqlx::Error> {
// CTE: Customer Lifetime Value
println!("\n--- Customer Lifetime Value (CTE) ---\n");
let results = sqlx::query_as::<_, CustomerLifetime>(
"WITH customer_orders AS (
SELECT
cu.name AS customer_name,
o.id AS order_id,
SUM(oi.quantity * oi.unit_price) AS order_total
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name, o.id
)
SELECT
customer_name,
SUM(order_total) AS total_spend,
COUNT(order_id) AS order_count,
ROUND(AVG(order_total), 2) AS avg_order_value
FROM customer_orders
GROUP BY customer_name
ORDER BY total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>8} {:>12}", "Customer", "Total", "Orders", "Avg Order");
println!("{}", "-".repeat(56));
for row in &results {
println!("{:<20} {:>12} {:>8} {:>12}", row.customer_name, row.total_spend, row.order_count, row.avg_order_value);
}
// NTILE: Customer Spending Tiers
println!("\n--- Customer Spending Tiers (NTILE) ---\n");
let results = sqlx::query_as::<_, SpendingTier>(
"WITH customer_totals AS (
SELECT
cu.name AS customer_name,
SUM(oi.quantity * oi.unit_price) AS total_spend
FROM customers cu
JOIN orders o ON o.customer_id = cu.id
JOIN order_items oi ON oi.order_id = o.id
GROUP BY cu.name
)
SELECT
customer_name,
total_spend,
NTILE(4) OVER (ORDER BY total_spend DESC) AS tier
FROM customer_totals
ORDER BY tier, total_spend DESC"
)
.fetch_all(pool)
.await?;
println!("{:<20} {:>12} {:>10}", "Customer", "Spend", "Tier");
println!("{}", "-".repeat(44));
for row in &results {
let tier_label = match row.tier {
1 => "Top 25%",
2 => "High",
3 => "Medium",
4 => "Low",
_ => "Unknown",
};
println!("{:<20} {:>12} {:>10}", row.customer_name, row.total_spend, tier_label);
}
// Moving Average with ROWS BETWEEN
println!("\n--- 3-Day Moving Average Revenue (ROWS BETWEEN) ---\n");
let results = sqlx::query_as::<_, MovingAvgRevenue>(
"WITH daily_rev AS (
SELECT
o.order_date::DATE AS order_date,
SUM(oi.quantity * oi.unit_price) AS daily_revenue
FROM orders o
JOIN order_items oi ON oi.order_id = o.id
GROUP BY o.order_date::DATE
)
SELECT
order_date,
daily_revenue,
ROUND(AVG(daily_revenue) OVER (
ORDER BY order_date
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
), 2) AS moving_avg_3day
FROM daily_rev
ORDER BY order_date"
)
.fetch_all(pool)
.await?;
println!("{:<12} {:>14} {:>16}", "Date", "Daily Rev", "3-Day Avg");
println!("{}", "-".repeat(44));
for row in &results {
let avg_str = match &row.moving_avg_3day {
Some(v) => format!("{}", v),
None => "N/A".to_string(),
};
println!("{:<12} {:>14} {:>16}", row.order_date, row.daily_revenue, avg_str);
}
Ok(())
}Your analytics tool now runs six distinct query patterns: GROUP BY aggregations, multi-column sorting, NULLS LAST handling, CTEs, NTILE window bucketing, and ROWS BETWEEN moving averages. That is the full base project complete.
Secret mission
Growth Analysis with LAG and ROLLUP
Your analytics tool shows snapshots of revenue and customer behaviour. But business stakeholders want trends. Can you add month-over-month growth tracking using the LAG window function, and a hierarchical revenue report using ROLLUP that shows category subtotals and a grand total in a single query?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Docker container pg-analytics running PostgreSQL 17 (uses minimal memory, no cost).
- Local project folder sales-analytics on your Desktop with Rust source code.
Keep everything running
No action needed. Choose this if you're still actively building or want to experiment with more SQL queries.
- The Docker container pg-analytics stays running in the background. Your seeded data remains intact.
- If you restart your computer, bring the container back up with docker start pg-analytics.
- Your sales-analytics folder remains on your Desktop, ready to edit and rerun.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your files and container data stay preserved.
- Stop the PostgreSQL container by running:
docker stop pg-analytics- Your data stays preserved inside the stopped container. When you're ready to continue, run docker start pg-analytics and then cargo run from the sales-analytics folder.
Delete - I don't want to use this again
Remove all project resources and start fresh. This is irreversible.
- Remove the Docker container and its data by running:
docker rm -f pg-analytics- Delete the project folder by running:
rm -rf ~/Desktop/sales-analytics- Verify everything is gone by running:
docker ps -a | grep pg-analytics
ls ~/Desktop/sales-analytics- Both commands should return no output, confirming the container and folder are removed.
Nice Work!
Nice Work!
Nice work! You just built a fully functional Rust CLI analytics tool that connects to a local PostgreSQL database and runs real-world analytical queries.
You've learned how to:
- Build a Rust CLI tool that connects to PostgreSQL via SQLx with async connection pooling, schema creation, and data seeding.
- Write advanced analytical SQL queries including GROUP BY aggregations, multi-column sorting with NULLS FIRST/LAST, and Common Table Expressions to break complex logic into readable steps.
- Apply window functions to solve real analytics problems, using NTILE to bucket customers into spending tiers and ROWS BETWEEN frame specifications for moving average calculations.
- Secret Mission: Implemented month-over-month revenue growth analysis with LAG and a hierarchical ROLLUP report with category and monthly subtotals.
Ready to quiz yourself?