Describe the result. Let the system plan the work.

SELECT madlib.logregr_train(
  'customer_features', 'churn_model_madlib',
  'churned',
  'ARRAY[1, recency_days, orders_90d, spend_90d]'
);

SELECT f.customer_id,
       madlib.logregr_predict_prob(
         m.coef,
         ARRAY[1, f.recency_days, f.orders_90d, f.spend_90d]
       ) AS churn_probability
FROM customer_features f, churn_model_madlib m;

SELECT * FROM pgml.train(
  project_name  => 'customer_churn',
  task          => 'classification',
  relation_name => 'customer_features',
  y_column_name => 'churned',
  algorithm     => 'xgboost',
  test_size     => 0.20
);

SELECT customer_id,
       pgml.predict(
         'customer_churn',
         ARRAY[recency_days, orders_90d, spend_90d]
       ) AS prediction
FROM customer_features;

CREATE MODEL churn_model
FROM analytics_db
  (SELECT recency_days, orders_90d,
          spend_90d, churned
   FROM customer_features)
PREDICT churned;

SELECT d.customer_id,
       m.churned AS predicted_churn,
       m.churned_confidence
FROM analytics_db.customer_features AS d
JOIN churn_model AS m;

CREATE TABLE churn_models
(
  name String,
  model AggregateFunction(
    stochasticLogisticRegression(0.01,0.0,10,'Adam'),
    UInt8, Float64, Float64, Float64)
) ENGINE=AggregatingMergeTree ORDER BY name;

INSERT INTO churn_models
SELECT 'churn_v1',
  stochasticLogisticRegressionState(0.01,0.0,10,'Adam')(
    churned, recency_days, orders_90d, spend_90d)
FROM customer_features;

WITH (SELECT model FROM churn_models WHERE name='churn_v1') AS m
SELECT customer_id,
  evalMLMethod(m, recency_days, orders_90d, spend_90d) AS score
FROM customer_features;

CREATE OR REPLACE TABLE train_snapshot AS
SELECT customer_id,
 date_diff('day',max(order_at),DATE '2026-08-31') AS recency_days,
 count(*) FILTER(WHERE order_at>=DATE '2026-06-01') AS orders_90d,
 sum(amount) FILTER(WHERE order_at>=DATE '2026-06-01') AS spend_90d,
 max(churned)::INTEGER AS churned
FROM read_parquet('orders/*.parquet')
GROUP BY customer_id;

COPY train_snapshot TO 'train_snapshot.parquet'
(FORMAT PARQUET, COMPRESSION ZSTD);