use cubeclient::models::{V1LoadRequestQuery, V1LoadRequestQueryTimeDimension};
use datafusion::{
    logical_plan::{JoinType, LogicalPlan, PlanVisitor},
    physical_plan::displayable,
    scalar::ScalarValue,
};
use pretty_assertions::assert_eq;
use regex::Regex;
use serde_json::json;
use std::sync::Arc;

use crate::{
    compile::{
        engine::df::scan::MemberField,
        rewrite::rewriter::Rewriter,
        test::{
            convert_select_to_query_plan, convert_select_to_query_plan_customized,
            convert_select_to_query_plan_with_config, init_testing_logger, LogicalPlanTestUtils,
            TestContext,
        },
        DatabaseProtocol,
    },
    config::ConfigObjImpl,
    transport::TransportLoadRequestQuery,
    CubeError,
};

#[tokio::test]
async fn test_simple_wrapper() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT COALESCE(customer_gender, 'N/A', 'NN'), AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("COALESCE"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, ROLLUP(2)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Rollup"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_with_aliases() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender as \"customer_gender1\", notes as \"notes\", AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY ROLLUP(1, 2)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Rollup"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_nested() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, avg(mp) from (SELECT customer_gender, notes, avg(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, 2) b GROUP BY ROLLUP(1, 2)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("ROLLUP(1, 2)"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_nested_from_asterisk() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, avg(avgPrice) from (SELECT * FROM KibanaSampleDataEcommerce) b GROUP BY ROLLUP(1, 2) ORDER BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Rollup"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_nested_with_aliases() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender as \"gender\", notes as \"notes\", avg(mp) from (SELECT customer_gender, notes, avg(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, 2) b GROUP BY ROLLUP(1, 2)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("ROLLUP(1, 2)"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_nested_complex() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, order_date, last_mod, avg(mp) from \
            (SELECT customer_gender, notes, order_date, last_mod, avg(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, 2, 3, 4) b \
            GROUP BY ROLLUP(1), ROLLUP(2), 3, CUBE(4)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("ROLLUP(1), ROLLUP(2), 3, CUBE(4)"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_placeholders() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY ROLLUP(1, 2)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Rollup"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_cube() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY CUBE(customer_gender, notes)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Cube"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_wrapper_group_by_rollup_complex() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, notes, has_subscription, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY ROLLUP(customer_gender, notes), has_subscription"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("Rollup"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_projection_empty_source() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (SELECT 'male' where 1  group by 'male' having 1 order by 'male' limit 1) as gender, avgPrice FROM KibanaSampleDataEcommerce a"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("(SELECT"));
    assert!(sql.contains("utf8__male__"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
    //println!("phys plan {:?}", physical_plan);
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_empty_source() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT avgPrice FROM KibanaSampleDataEcommerce a where customer_gender = (SELECT 'male' )"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("(SELECT"));
    assert!(sql.contains("utf8__male__"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
    //println!("phys plan {:?}", physical_plan);
}

#[tokio::test]
async fn test_wrapper_filter_subquery_current_timestamp() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        r#"
        SELECT customer_gender, AVG(avgPrice) mp
        FROM KibanaSampleDataEcommerce a
        WHERE order_date >= (SELECT DATE_TRUNC('year', DATE_TRUNC('day', CURRENT_TIMESTAMP)))
        GROUP BY 1
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("NOW() AT TIME ZONE 'UTC'"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_projection_aggregate_empty_source() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (SELECT 'male'), avg(avgPrice) FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("(SELECT"));
    assert!(sql.contains("utf8__male__"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_in_empty_source() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, avgPrice FROM KibanaSampleDataEcommerce a where customer_gender in (select 'male')"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("IN (SELECT"));
    assert!(sql.contains("utf8__male__"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_and_projection_empty_source() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (select 'male'), avgPrice FROM KibanaSampleDataEcommerce a where customer_gender in (select 'female')"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();

    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("IN (SELECT"));
    assert!(sql.contains("(SELECT"));
    assert!(sql.contains("utf8__male__"));
    assert!(sql.contains("utf8__female__"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_projection() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (SELECT customer_gender FROM KibanaSampleDataEcommerce LIMIT 1) as gender, avgPrice FROM KibanaSampleDataEcommerce a"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("(SELECT"));
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\\\\\\\"limit\\\\\\\": 1"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_projection_aggregate() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (SELECT customer_gender FROM KibanaSampleDataEcommerce WHERE customer_gender = 'male' LIMIT 1), avg(avgPrice) FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("(SELECT"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_equal() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, avgPrice FROM KibanaSampleDataEcommerce a where customer_gender = (select customer_gender from KibanaSampleDataEcommerce limit 1)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("(SELECT"));
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\\\\\\\"limit\\\\\\\": 1"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_in() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT customer_gender, avgPrice FROM KibanaSampleDataEcommerce a where customer_gender in (select customer_gender from KibanaSampleDataEcommerce)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("IN (SELECT"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

#[tokio::test]
async fn test_simple_subquery_wrapper_filter_and_projection() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT (select customer_gender from KibanaSampleDataEcommerce limit 1), avgPrice FROM KibanaSampleDataEcommerce a where customer_gender in (select customer_gender from KibanaSampleDataEcommerce)"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();

    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("IN (SELECT"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

// TODO add more time zones
// TODO add more TS syntax variants
// TODO add TIMESTAMPTZ variant
/// Using TIMESTAMP WITH TIME ZONE with actual timezone in wrapper should render proper timestamptz in SQL
#[tokio::test]
async fn test_wrapper_timestamptz() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    customer_gender
FROM KibanaSampleDataEcommerce
WHERE
    order_date >= TIMESTAMP WITH TIME ZONE '2024-02-03T04:05:06Z'
    AND
--   This filter should trigger pushdown
    LOWER(customer_gender) = 'male'
GROUP BY
    1
;
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains(
            "${KibanaSampleDataEcommerce.order_date} >= timestamptz '2024-02-03T04:05:06.000Z'"
        ));
}

// TODO add more time zones
// TODO add more TS syntax variants
// TODO add TIMESTAMPTZ variant
/// Using TIMESTAMP WITH TIME ZONE with actual timezone in ungrouped wrapper should render proper timestamptz in SQL
#[tokio::test]
async fn test_wrapper_timestamptz_ungrouped() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    customer_gender
FROM KibanaSampleDataEcommerce
WHERE
    order_date >= TIMESTAMP WITH TIME ZONE '2024-02-03T04:05:06Z'
    AND
--   This filter should trigger pushdown
    LOWER(customer_gender) = 'male'
;
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains(
            "${KibanaSampleDataEcommerce.order_date} >= timestamptz '2024-02-03T04:05:06.000Z'"
        ));
}

/// Using NOW() in wrapper should render NOW() in SQL
#[tokio::test]
async fn test_wrapper_now() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    customer_gender
FROM KibanaSampleDataEcommerce
WHERE
    order_date >= NOW()
    AND
--   This filter should trigger pushdown
    LOWER(customer_gender) = 'male'
GROUP BY
    1
;
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("${KibanaSampleDataEcommerce.order_date} >= NOW()"));
}

/// Using NOW() in ungrouped wrapper should render NOW() in SQL
#[tokio::test]
async fn test_wrapper_now_ungrouped() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    customer_gender
FROM KibanaSampleDataEcommerce
WHERE
    order_date >= NOW()
    AND
--   This filter should trigger pushdown
    LOWER(customer_gender) = 'male'
;
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("${KibanaSampleDataEcommerce.order_date} >= NOW()"));
}

#[tokio::test]
async fn test_case_wrapper() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN COALESCE(customer_gender, 'N/A', 'NN') = 'female' THEN 'f' ELSE 'm' END, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_distinct() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        r#"SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, COUNT(DISTINCT countDistinct) mp
            FROM KibanaSampleDataEcommerce a
            WHERE
              (
                (
                  ( a.order_date ) >= '2024-01-01'
                  AND ( a.order_date ) < '2024-02-01'
                )
              )
            GROUP BY 1"#
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_alias_with_order() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END AS \"f822c516-3515-11c2-8464-5d4845a02f73\", AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END ORDER BY CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END NULLS FIRST LIMIT 500"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("ORDER BY \"a\".\"case_when_a_cust\""));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_ungrouped() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_non_strict_match() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let mut config = ConfigObjImpl::default();

    config.disable_strict_agg_type_match = true;

    let query_plan = convert_select_to_query_plan_with_config(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, SUM(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
        Arc::new(config)
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_ungrouped_sorted() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1 ORDER BY 1 DESC"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("ORDER BY"));
}

#[tokio::test]
async fn test_case_wrapper_ungrouped_sorted_aliased() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT x FROM (SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END x, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1 ORDER BY 1 DESC) b"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        // TODO test without depend on column name
        .contains("ORDER BY \"a\".\"case_when"));
}

#[tokio::test]
async fn test_case_wrapper_with_internal_limit() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1 LIMIT 1123"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    assert!(
        logical_plan
            .find_cube_scan_wrapped_sql()
            .wrapped_sql
            .sql
            .contains("1123"),
        "SQL contains 1123: {}",
        logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql
    );

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_with_system_fields() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, __user, __cubeJoinField, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, 2, 3 LIMIT 1123"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();

    assert!(
        logical_plan
            .find_cube_scan_wrapped_sql()
            .wrapped_sql
            .sql
            .contains(
                "\\\"cubeName\\\":\\\"KibanaSampleDataEcommerce\\\",\\\"alias\\\":\\\"__user\\\""
            ),
        r#"SQL contains `\"cubeName\":\"KibanaSampleDataEcommerce\",\"alias\":\"__user\"` {}"#,
        logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql
    );

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_with_limit() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT * FROM (SELECT CASE WHEN customer_gender = 'female' THEN 'f' ELSE 'm' END, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1) q LIMIT 1123"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    assert!(
        logical_plan
            .find_cube_scan_wrapped_sql()
            .wrapped_sql
            .sql
            .contains("1123"),
        "SQL contains 1123: {}",
        logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql
    );

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_with_null() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN taxful_total_price IS NULL THEN NULL WHEN taxful_total_price < taxful_total_price * 2 THEN COALESCE(taxful_total_price, 0, 0) END, AVG(avgPrice) FROM KibanaSampleDataEcommerce GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("CASE WHEN"));

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_ungrouped_on_dimension() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT CASE WHEN SUM(taxful_total_price) > 0 THEN SUM(taxful_total_price) ELSE 0 END FROM KibanaSampleDataEcommerce a"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
        .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

#[tokio::test]
async fn test_case_wrapper_escaping() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan_customized(
        "SELECT CASE WHEN customer_gender = '\\`' THEN COALESCE(customer_gender, 'N/A', 'NN') ELSE 'N/A' END as \"\\`\", AVG(avgPrice) FROM KibanaSampleDataEcommerce a GROUP BY 1".to_string(),
        DatabaseProtocol::PostgreSQL,
        vec![
            ("expressions/binary".to_string(), "{{ left }} \\`{{ op }} {{ right }}".to_string())
        ],
    ).await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        // Expect 6 backslashes as output is JSON and it's escaped one more time
        .contains("\\\\\\\\\\\\`"));
}

#[tokio::test]
async fn test_wrapper_ilike_lowered_pushdown() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan_customized(
        "SELECT customer_gender ILIKE 'fem' AS g, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
        vec![(
            "expressions/ilike".to_string(),
            "LOWER({{ expr }}) {% if negated %}NOT {% endif %}LIKE LOWER({{ pattern }})".to_string(),
        )],
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains(") LIKE LOWER("),
        "expected lowered ILIKE pushdown (LOWER(..) LIKE LOWER(..)), got: {}",
        sql
    );

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

/// The TO_CHAR format string is passed to the dialect template as a bound parameter,
/// not as an inline literal, so dialects that need to translate PostgreSQL format tokens
/// (e.g. Databricks translating `YYYY-MM-DD` to Spark's `yyyy-MM-dd`) must do so in SQL,
/// wrapping the parameter itself. The template below is a copy of the Databricks one
/// (see `DatabricksQuery.sqlTemplates` in `@cubejs-backend/databricks-jdbc-driver`);
/// the REPLACE chain is order-sensitive, so the translations it produces are verified
/// here by simulating the sequential replace-all semantics of SQL REPLACE
#[tokio::test]
async fn test_wrapper_to_char_format_as_param() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    const TO_CHAR_TEMPLATE: &str = "TO_CHAR({{ args[0] }}, REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE(REPLACE({{ args[1] }}, 'HH12', 'hh'), 'HH24', '@H24@'), 'HH', 'hh'), '@H24@', 'HH'), 'MI', 'mm'), 'SS', 'ss'), 'MS', 'SSS'), 'US', 'SSSSSS'), 'YYYY', 'yyyy'), 'YY', 'yy'), 'Month', 'MMMM'), 'Mon', 'MMM'), 'DDD', '@DOY@'), 'Day', 'EEEE'), 'Dy', 'EEE'), 'DD', 'dd'), '@DOY@', 'DDD'), 'AM', 'a'), 'PM', 'a'))";

    let query_plan = convert_select_to_query_plan_customized(
        "SELECT TO_CHAR(DATE_TRUNC('week', order_date), 'YYYY-MM-DD') d1, TO_CHAR(DATE_TRUNC('day', order_date), 'HH24:MI:SS') d2, AVG(avgPrice) mp FROM KibanaSampleDataEcommerce a GROUP BY 1, 2"
            .to_string(),
        DatabaseProtocol::PostgreSQL,
        vec![(
            "functions/TO_CHAR".to_string(),
            TO_CHAR_TEMPLATE.to_string(),
        )],
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let wrapped_sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql;
    assert!(
        wrapped_sql.sql.contains("TO_CHAR(")
            && wrapped_sql.sql.contains("'HH12', 'hh')")
            && wrapped_sql.sql.contains("'PM', 'a'))"),
        "expected TO_CHAR template with the REPLACE translation chain applied, got: {}",
        wrapped_sql.sql
    );
    for format in ["YYYY-MM-DD", "HH24:MI:SS"] {
        assert!(
            wrapped_sql.values.contains(&Some(format.to_string())),
            "expected format string '{}' to be passed as a bound parameter, got: {:?}",
            format,
            wrapped_sql.values
        );
    }

    // Extract the ordered (from, to) REPLACE pairs from the template and verify that
    // applying them sequentially translates PostgreSQL formats to correct Spark
    // datetime patterns; expected values were validated against Databricks itself
    let pair_regex = Regex::new(r"'([^']+)', '([^']+)'\)").unwrap();
    let replacements = pair_regex
        .captures_iter(TO_CHAR_TEMPLATE)
        .map(|c| (c[1].to_string(), c[2].to_string()))
        .collect::<Vec<_>>();
    assert_eq!(replacements.len(), 19);

    let translate = |format: &str| {
        replacements
            .iter()
            .fold(format.to_string(), |acc, (from, to)| acc.replace(from, to))
    };

    for (postgres_format, spark_pattern) in [
        ("YYYY-MM-DD", "yyyy-MM-dd"),
        ("YYYY-MM-DD HH24:MI:SS", "yyyy-MM-dd HH:mm:ss"),
        ("YYYY-MM-DD HH24:MI:SS.MS", "yyyy-MM-dd HH:mm:ss.SSS"),
        ("YYYY-MM-DD HH24:MI:SS.US", "yyyy-MM-dd HH:mm:ss.SSSSSS"),
        // Bare HH is a 12-hour clock in PostgreSQL while Spark's HH is 24-hour
        ("HH:MI:SS AM", "hh:mm:ss a"),
        ("HH12:MI PM", "hh:mm a"),
        ("HH24:MI:SS", "HH:mm:ss"),
        // DDD (day-of-year) must not be clobbered by the DD (day-of-month) rule
        ("DDD", "DDD"),
        ("YYYY-DDD", "yyyy-DDD"),
        ("Day", "EEEE"),
        ("Dy", "EEE"),
        ("Month", "MMMM"),
        ("Mon", "MMM"),
        ("YYYY-Mon", "yyyy-MMM"),
        ("MMDD", "MMdd"),
        ("YY-MM", "yy-MM"),
    ] {
        let translated = translate(postgres_format);
        assert_eq!(
            translated, spark_pattern,
            "expected '{}' to translate to '{}'",
            postgres_format, spark_pattern
        );
        assert!(
            !translated.contains('@'),
            "expected no stash placeholders left in '{}'",
            translated
        );
    }

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

/// Test aliases for grouped CubeScan in wrapper
/// qualifiers from join should get remapped to single from alias
/// long generated aliases from Datafusion should get shortened
#[tokio::test]
async fn test_join_wrapper_cubescan_aliasing() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
WITH
-- This subquery should be represented as CubeScan(ungrouped=false) inside CubeScanWrapper
cube_scan_subq AS (
    SELECT
        logs_alias.content logs_content,
        DATE_TRUNC('month', kibana_alias.last_mod) last_mod_month,
        kibana_alias.__user AS cube_user,
        1 AS literal,
        -- Columns without aliases should also work
        DATE_TRUNC('month', kibana_alias.order_date),
        kibana_alias.__cubeJoinField,
        2,
        CASE
            WHEN sum(kibana_alias."sumPrice") IS NOT NULL
                THEN sum(kibana_alias."sumPrice")
            ELSE 0
            END sum_price
    FROM KibanaSampleDataEcommerce kibana_alias
    JOIN Logs logs_alias
    ON kibana_alias.__cubeJoinField = logs_alias.__cubeJoinField
    GROUP BY 1,2,3,4,5,6,7
),
filter_subq AS (
    SELECT
        Logs.content logs_content_filter
    FROM Logs
    GROUP BY
        logs_content_filter
)
SELECT
    -- Should use SELECT * here to reference columns without aliases.
    -- But it's broken ATM in DF, initial plan contains `Projection: ... #__subquery-0.logs_content_filter` on top, but it should not be there
    -- TODO fix it
    logs_content,
    cube_user,
    literal
FROM cube_scan_subq
WHERE
    -- This subquery filter should trigger wrapping of whole query
    logs_content IN (
        SELECT
            logs_content_filter
        FROM filter_subq
    )
;
"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;

    assert_eq!(
        logical_plan.find_cube_scan().request,
        V1LoadRequestQuery {
            measures: Some(vec!["KibanaSampleDataEcommerce.sumPrice".to_string(),]),
            dimensions: Some(vec!["Logs.content".to_string(),]),
            time_dimensions: Some(vec![
                V1LoadRequestQueryTimeDimension {
                    dimension: "KibanaSampleDataEcommerce.last_mod".to_string(),
                    granularity: Some("month".to_string()),
                    date_range: None,
                },
                V1LoadRequestQueryTimeDimension {
                    dimension: "KibanaSampleDataEcommerce.order_date".to_string(),
                    granularity: Some("month".to_string()),
                    date_range: None,
                },
            ]),
            segments: Some(vec![]),
            order: Some(vec![]),
            join_hints: Some(vec![vec![
                "KibanaSampleDataEcommerce".to_string(),
                "Logs".to_string(),
            ],]),
            ..Default::default()
        }
    );

    assert_eq!(
        logical_plan.find_cube_scan().member_fields,
        vec![
            MemberField::regular("Logs.content".to_string()),
            MemberField::time_dimension(
                "KibanaSampleDataEcommerce.last_mod".to_string(),
                "month".to_string()
            ),
            MemberField::Literal(ScalarValue::Utf8(None)),
            MemberField::Literal(ScalarValue::Int64(Some(1))),
            MemberField::time_dimension(
                "KibanaSampleDataEcommerce.order_date".to_string(),
                "month".to_string()
            ),
            MemberField::Literal(ScalarValue::Utf8(None)),
            MemberField::Literal(ScalarValue::Int64(Some(2))),
            MemberField::regular("KibanaSampleDataEcommerce.sumPrice".to_string()),
        ],
    );

    // Check that all aliases from different tables have same qualifier, and that names are simple and short
    // logs_content => logs_alias.content
    // last_mod_month => DATE_TRUNC('month', kibana_alias.last_mod),
    // sum_price => CASE WHEN sum(kibana_alias."sumPrice") ... END
    let content_re = Regex::new(r#""logs_alias"."[a-zA-Z0-9_]{1,16}" "logs_content""#).unwrap();
    assert!(content_re.is_match(&sql));
    let last_mod_month_re =
        Regex::new(r#""logs_alias"."[a-zA-Z0-9_]{1,16}" "last_mod_month""#).unwrap();
    assert!(last_mod_month_re.is_match(&sql));
    let sum_price_re = Regex::new(r#"CASE WHEN \("logs_alias"."[a-zA-Z0-9_]{1,16}" IS NOT NULL\) THEN "logs_alias"."[a-zA-Z0-9_]{1,16}" ELSE 0 END "sum_price""#)
        .unwrap();
    assert!(sum_price_re.is_match(&sql));
    let cube_user_re = Regex::new(r#""logs_alias"."[a-zA-Z0-9_]{1,16}" "cube_user""#).unwrap();
    assert!(cube_user_re.is_match(&sql));
    let literal_re = Regex::new(r#""logs_alias"."[a-zA-Z0-9_]{1,16}" "literal""#).unwrap();
    assert!(literal_re.is_match(&sql));
}

/// Regression test for grouped-grouped joins with non-Inner/Left join types under SQL
/// push down. Both sides are grouped subqueries joined with FULL OUTER JOIN (the shape
/// produced by period-over-period queries). Before `join_types/full` was wired up, this
/// failed with `Unsupported join type for join subquery: Full`.
#[tokio::test]
async fn test_grouped_join_wrapper_full_outer() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        WITH first_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS first_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        ),
        last_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS last_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        )
        SELECT
            COALESCE(f.customer_gender, l.customer_gender) AS customer_gender,
            COALESCE(f.first_price, 0) AS first_price,
            COALESCE(l.last_price, 0) AS last_price
        FROM first_period f
        FULL OUTER JOIN last_period l ON f.customer_gender = l.customer_gender
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    // Whole query must be pushed down to a single wrapped SQL with a FULL JOIN
    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("FULL JOIN"),
        "wrapped SQL is missing FULL JOIN:\n{}",
        sql
    );
}

/// Same as [`test_grouped_join_wrapper_full_outer`], but for RIGHT JOIN.
#[tokio::test]
async fn test_grouped_join_wrapper_right() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        WITH first_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS first_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        ),
        last_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS last_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        )
        SELECT
            COALESCE(f.customer_gender, l.customer_gender) AS customer_gender,
            COALESCE(f.first_price, 0) AS first_price,
            COALESCE(l.last_price, 0) AS last_price
        FROM first_period f
        RIGHT JOIN last_period l ON f.customer_gender = l.customer_gender
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    // Whole query must be pushed down to a single wrapped SQL with a RIGHT JOIN
    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("RIGHT JOIN"),
        "wrapped SQL is missing RIGHT JOIN:\n{}",
        sql
    );
}

/// When the data source has no `join_types/full` template, a grouped-grouped FULL JOIN
/// must not be pushed down: the join stays in the DataFusion plan and the query is still
/// plannable instead of failing with `Unsupported join type for join subquery`.
#[tokio::test]
async fn test_grouped_join_wrapper_full_outer_without_template() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan_customized(
        // language=PostgreSQL
        r#"
        WITH first_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS first_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        ),
        last_period AS (
            SELECT
                customer_gender,
                sum(sumPrice) AS last_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        )
        SELECT
            COALESCE(f.customer_gender, l.customer_gender) AS customer_gender,
            COALESCE(f.first_price, 0) AS first_price,
            COALESCE(l.last_price, 0) AS last_price
        FROM first_period f
        FULL OUTER JOIN last_period l ON f.customer_gender = l.customer_gender
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
        // Emulate a data source without FULL JOIN support: empty value removes the template
        vec![("join_types/full".to_string(), "".to_string())],
    )
    .await;

    // Query must still be plannable, with the join executed by DataFusion
    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    struct FindJoinVisitor(Option<JoinType>);

    impl PlanVisitor for FindJoinVisitor {
        type Error = CubeError;

        fn pre_visit(&mut self, plan: &LogicalPlan) -> Result<bool, Self::Error> {
            if let LogicalPlan::Join(join) = plan {
                self.0 = Some(join.join_type);
            }
            Ok(true)
        }
    }

    let logical_plan = query_plan.as_logical_plan();
    let mut visitor = FindJoinVisitor(None);
    logical_plan.accept(&mut visitor).unwrap();
    assert_eq!(
        visitor.0,
        Some(JoinType::Full),
        "expected FULL JOIN to stay in the DataFusion plan, got:\n{}",
        logical_plan.display_indent(),
    );
}

/// FULL OUTER JOIN between grouped CTEs from two *different* cubes (no join relationship
/// defined between them). Both sides are standalone grouped subqueries joined in SQL
/// (non-push-to-Cube path), so FULL should be supported and pushed down.
#[tokio::test]
async fn test_grouped_join_wrapper_full_outer_cross_cube() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        WITH m1 AS (
            SELECT
                id,
                sum(sumPrice) AS first_price
            FROM KibanaSampleDataEcommerce
            GROUP BY 1
        ),
        m2 AS (
            SELECT
                dim0,
                MEASURE(WideCube.count) AS cnt
            FROM WideCube
            GROUP BY 1
        )
        SELECT
            COALESCE(m1.id, m2.dim0) AS id,
            m1.first_price,
            m2.cnt
        FROM m1
        FULL OUTER JOIN m2 ON m1.id = m2.dim0
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    // Whole query must be pushed down to a single wrapped SQL with a FULL JOIN
    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("FULL JOIN"),
        "wrapped SQL is missing FULL JOIN:\n{}",
        sql
    );
}

/// LEFT JOIN between grouped CTEs from two *different* cubes (no join relationship
/// defined between them), joined on multiple dimensions sharing the same output names.
#[tokio::test]
async fn test_grouped_join_wrapper_left_cross_cube_multi_condition() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        WITH m1 AS (
            SELECT
                customer_gender,
                notes,
                sum(sumPrice) AS sum_price
            FROM KibanaSampleDataEcommerce
            WHERE
                has_subscription = true
                AND customer_gender IS NOT NULL
                AND order_date BETWEEN '2024-01-01T00:00:00.000' AND '2024-01-31T23:59:59.999'
            GROUP BY 1, 2
        ),
        m2 AS (
            SELECT
                dim0 AS customer_gender,
                dim1 AS notes,
                MEASURE(WideCube.count) AS cnt
            FROM WideCube
            WHERE
                dim2 IN ('female')
                AND dim3 IS NOT NULL
            GROUP BY 1, 2
        )
        SELECT
            COALESCE(m1.customer_gender, m2.customer_gender) AS customer_gender,
            COALESCE(m1.notes, m2.notes) AS notes,
            m1.sum_price,
            m2.cnt
        FROM m1
        LEFT JOIN m2 ON m1.customer_gender = m2.customer_gender AND m1.notes = m2.notes
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    // Whole query must be pushed down to a single wrapped SQL with a LEFT JOIN
    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("LEFT JOIN"),
        "wrapped SQL is missing LEFT JOIN:\n{}",
        sql
    );
    // Same-named columns from both join sides must keep distinct aliases in the outer
    // projection: COALESCE over both sides must not reference the same (right-side) column
    // twice, or LEFT JOIN rows without a match would coalesce to NULL.
    assert!(
        sql.contains(r#"COALESCE("m1"."customer_gender", "m1"."customer_gender_1")"#),
        "left side of COALESCE was remapped to the right-side column:\n{}",
        sql
    );
    assert!(
        sql.contains(r#"COALESCE("m1"."notes", "m1"."notes_1")"#),
        "left side of COALESCE was remapped to the right-side column:\n{}",
        sql
    );
}

/// Regression test for smoke test "select __user and literal grouped under wrapper".
/// Inner CTE has unaliased DATE_TRUNC expressions; outer query references those columns
/// through the SubqueryAlias qualifier (`cube_scan_subq`). The CTE-level Remapper must
/// publish a mapping under the SubqueryAlias qualifier — otherwise the outer projection
/// cannot remap `cube_scan_subq.datetrunc(Utf8("day"),...)` to the short alias, and the
/// DataFusion-internal expression name leaks into the generated SQL.
#[tokio::test]
async fn test_single_cube_grouped_wrapper_with_join_field() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
WITH
cube_scan_subq AS (
    SELECT
        customer_gender AS my_gender,
        DATE_TRUNC('month', order_date) AS my_order_month,
        __user AS my_user,
        1 AS my_literal,
        id,
        DATE_TRUNC('day', order_date),
        __cubeJoinField,
        2
    FROM KibanaSampleDataEcommerce
    GROUP BY 1,2,3,4,5,6,7,8
),
filter_subq AS (
    SELECT
        customer_gender gender_filter
    FROM KibanaSampleDataEcommerce
    GROUP BY
        gender_filter
)
SELECT
    my_order_month,
    my_gender,
    my_user,
    my_literal
FROM cube_scan_subq
WHERE
    my_gender IN (
        SELECT
            gender_filter
        FROM filter_subq
    )
GROUP BY 1,2,3,4
ORDER BY 1,2,3,4
;
"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        !sql.contains("datetrunc(Utf8"),
        "wrapped SQL leaked DataFusion-rendered expr name (e.g. datetrunc(Utf8(\"day\"),...)):\n{}",
        sql
    );
}

/// Regression test for grouped-grouped join SQL push down (Tableau-style query).
/// Join condition columns must be remapped to generated aliases of the joined
/// subqueries — otherwise the generated ON clause references the original column
/// name (e.g. `"t0"."My Notes"`) which does not exist in the aliased subqueries.
#[tokio::test]
async fn test_wrapper_grouped_join_grouped_condition_remapping() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT "t0"."Customer Gender" AS "Customer Gender",
    SUM("t2"."__measure__1") AS "sum:measure:ok"
FROM (
    SELECT
        customer_gender AS "Customer Gender",
        notes AS "My Notes"
    FROM KibanaSampleDataEcommerce
    GROUP BY 1, 2
) "t0"
INNER JOIN (
    SELECT
        notes AS "My Notes",
        AVG(avgPrice) AS "__measure__1"
    FROM KibanaSampleDataEcommerce
    GROUP BY 1
) "t2" ON ("t0"."My Notes" = "t2"."My Notes")
GROUP BY 1
;
"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(sql.contains("INNER JOIN ("));
    assert!(
        !sql.contains(r#""My Notes""#),
        "join condition leaked original column name instead of remapped alias:\n{}",
        sql
    );
}

/// Regression test for window expression on top of a grouped join subquery
/// (Sigma-style query). Window expr rebase in converter must keep field names
/// consistent with column references in plans above.
#[tokio::test]
async fn test_wrapper_window_over_grouped_join() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        WITH
        "qt_0" AS (
            SELECT
                "ta_1".content "ca_1",
                DATE_TRUNC('month', "ta_2".order_date) "ca_2",
                CASE
                    WHEN sum("ta_2"."sumPrice") IS NOT NULL THEN sum("ta_2"."sumPrice")
                    ELSE 0
                END "ca_3"
            FROM KibanaSampleDataEcommerce "ta_2"
            JOIN Logs "ta_1"
                ON "ta_2".__cubeJoinField = "ta_1".__cubeJoinField
            GROUP BY
                "ca_1",
                "ca_2"
        ),
        "qt_1" AS (
            SELECT
                RANK() OVER (
                    PARTITION BY DATE_TRUNC('month', "ta_3"."ca_2")
                    ORDER BY
                        DATE_TRUNC('month', "ta_3"."ca_2"),
                        CASE
                            WHEN sum("ta_3"."ca_3") IS NOT NULL THEN sum("ta_3"."ca_3")
                            ELSE 0
                        END DESC,
                        "ta_3"."ca_1"
                ) "ca_4",
                DATE_TRUNC('month', "ta_3"."ca_2") "ca_5",
                "ta_3"."ca_1" "ca_6",
                CASE
                    WHEN sum("ta_3"."ca_3") IS NOT NULL THEN sum("ta_3"."ca_3")
                    ELSE 0
                END "ca_7"
            FROM "qt_0" "ta_3"
            GROUP BY
                "ca_5",
                "ca_6"
        )
        SELECT
            "ta_4"."ca_5" "ca_8",
            "ta_4"."ca_6" "ca_9",
            "ta_4"."ca_7" "ca_10"
        FROM "qt_1" "ta_4"
        WHERE "ta_4"."ca_4" <= 1
        ORDER BY
            "ca_8" ASC,
            "ca_10" DESC,
            "ca_9" ASC
        ;
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

/// Test that WrappedSelect(... limit=Some(0) ...) will render it correctly
#[tokio::test]
async fn test_wrapper_limit_zero() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
            SELECT
                MIN(t.a)
            FROM (
                SELECT
                    MAX(order_date) AS a
                FROM
                    KibanaSampleDataEcommerce
                LIMIT 10
            ) t LIMIT 0
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("LIMIT 0"));

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

/// Tests that Aggregation(Filter(CubeScan(ungrouped=true))) with expresions in filter
/// can be executed as a single ungrouped=false load query
#[tokio::test]
async fn test_wrapper_filter_flatten() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
            SELECT
                customer_gender,
                SUM(sumPrice)
            FROM
                KibanaSampleDataEcommerce
            WHERE
                LOWER(customer_gender) = 'male'
            GROUP BY
                1
            "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert_eq!(
        query_plan
            .as_logical_plan()
            .find_cube_scan_wrapped_sql()
            .request,
        TransportLoadRequestQuery {
            measures: Some(vec![json!({
                "cubeName": "KibanaSampleDataEcommerce",
                "alias": "sum_kibanasample",
                "expr": {
                    "type": "SqlFunction",
                    "cubeParams": ["KibanaSampleDataEcommerce"],
                    // This is grouped query, KibanaSampleDataEcommerce.sumPrice is correct in this context
                    // SUM(sumPrice) will be incrrect here, it would lead to SUM(SUM(sql)) in generated query
                    "sql": "${KibanaSampleDataEcommerce.sumPrice}",
                },
                "groupingSet": null,
            })
            .to_string(),]),
            dimensions: Some(vec![json!({
                "cubeName": "KibanaSampleDataEcommerce",
                "alias": "customer_gender",
                "expr": {
                    "type": "SqlFunction",
                    "cubeParams": ["KibanaSampleDataEcommerce"],
                    "sql": "${KibanaSampleDataEcommerce.customer_gender}",
                },
                "groupingSet": null,
            })
            .to_string(),]),
            segments: Some(vec![json!({
                "cubeName": "KibanaSampleDataEcommerce",
                "alias": "lower_kibanasamp",
                "expr": {
                    "type": "SqlFunction",
                    "cubeParams": ["KibanaSampleDataEcommerce"],
                    "sql": "(LOWER(${KibanaSampleDataEcommerce.customer_gender}) = $0$)",
                },
                "groupingSet": null,
            })
            .to_string(),]),
            time_dimensions: None,
            order: Some(vec![]),
            limit: Some(50000),
            ..Default::default()
        }
    );
}

/// Regular aggregation over CubeScan(limit=n, ungrouped=true) is NOT pushed to CubeScan
/// and inner ungrouped CubeScan should have both proper members and limit
#[tokio::test]
async fn wrapper_agg_over_limit() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            customer_gender
        FROM (
            SELECT
                customer_gender
            FROM
                KibanaSampleDataEcommerce
            LIMIT 5
        ) scan
        GROUP BY
            1
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert_eq!(
        logical_plan.find_cube_scan().request,
        V1LoadRequestQuery {
            measures: Some(vec![]),
            dimensions: Some(vec![
                "KibanaSampleDataEcommerce.customer_gender".to_string(),
            ]),
            segments: Some(vec![]),
            order: Some(vec![]),
            limit: Some(5),
            ungrouped: Some(true),
            ..Default::default()
        }
    );

    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\"limit\": 5"));
    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\"ungrouped\": true"));
}

/// Aggregation(dimension) over CubeScan(limit=n, ungrouped=true) is NOT pushed to CubeScan
/// and inner ungrouped CubeScan should have both proper members and limit
#[tokio::test]
async fn wrapper_agg_dimension_over_limit() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            MAX(customer_gender)
        FROM (
            SELECT
                customer_gender
            FROM
                KibanaSampleDataEcommerce
            LIMIT 5
        ) scan
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert_eq!(
        logical_plan.find_cube_scan().request,
        V1LoadRequestQuery {
            measures: Some(vec![]),
            dimensions: Some(vec![
                "KibanaSampleDataEcommerce.customer_gender".to_string(),
            ]),
            segments: Some(vec![]),
            order: Some(vec![]),
            limit: Some(5),
            ungrouped: Some(true),
            ..Default::default()
        }
    );

    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\"limit\": 5"));
    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("\"ungrouped\": true"));
}

// TODO allow number measures and add test for those
/// Projection(Filter(CubeScan(ungrouped))) should have projection expressions pushed down to Cube
#[tokio::test]
async fn wrapper_projection_flatten_simple_measure() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
  maxPrice
FROM
  MultiTypeCube
WHERE
  LOWER(CAST(dim_num0 AS TEXT)) = 'all'
;
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let request = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .request;
    assert_eq!(request.measures.unwrap().len(), 1);
    assert_eq!(request.dimensions.unwrap().len(), 0);
}

#[tokio::test]
async fn wrapper_duplicated_members() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        format!(
            r#"
SELECT
    "foo",
    "bar",
    CASE
        WHEN "bar" IS NOT NULL
        THEN 1
        ELSE 0
        END
    AS "bar_expr"
FROM (
    SELECT
        "rows"."foo" AS "foo",
        "rows"."bar" AS "bar"
    FROM (
        SELECT
            "dim_str0" AS "foo",
            "dim_str0" AS "bar"
        FROM MultiTypeCube
    ) "rows"
    GROUP BY
        "foo",
        "bar"
) "_"
ORDER BY
    "bar_expr"
LIMIT 1
;
        "#
        )
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    // Generated SQL should contain realiasing of one member to two columns
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains(r#""foo" "foo""#));
    assert!(logical_plan
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains(r#""foo" "bar""#));
}

/// Simple wrapper with cast should have explicit members, not zero
#[tokio::test]
async fn wrapper_cast_limit_explicit_members() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            CAST(dim_date0 AS DATE) AS "dim_date0"
        FROM
            MultiTypeCube
        LIMIT 10
        ;
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    // Query should mention just a single member
    let request = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .request;
    assert_eq!(request.measures.unwrap().len(), 0);
    assert_eq!(request.dimensions.unwrap().len(), 1);
}

#[tokio::test]
async fn wrapper_typed_null() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            dim_str0,
            AVG(avgPrice),
            CASE
                WHEN SUM((NULLIF(0.0, 0.0))) IS NOT NULL THEN SUM((NULLIF(0.0, 0.0)))
                ELSE 0
                END
        FROM MultiTypeCube
        GROUP BY 1
        ;"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("SUM(CAST(NULL AS DOUBLE))"));
}

/// Tests that exactly same expression in projection and filter have correct alias after rewriting
#[tokio::test]
async fn test_same_expression_in_projection_and_filter() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    DATE_TRUNC('day', CAST(dim_date0 AS TIMESTAMP))
FROM MultiTypeCube
WHERE
    DATE_TRUNC('day', CAST(dim_date0 AS TIMESTAMP)) >=
     '2025-01-01'
    AND LOWER(dim_str0) = 'some value'
GROUP BY
    1
;
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let request = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .request;
    let dimensions = request.dimensions.unwrap();
    assert_eq!(dimensions.len(), 1);
    let dimension = &dimensions[0];
    assert!(dimension.contains(".day"));
    let segments = request.segments.unwrap();
    assert_eq!(segments.len(), 1);
    let segment = &segments[0];
    assert!(segment.contains(".day"));
}

/// Aggregation with falsy filter should NOT get pushed to CubeScan with limit=0
/// This test currently produces WrappedSelect with WHERE FALSE, which is OK for our purposes
#[tokio::test]
async fn select_agg_where_false() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        "SELECT SUM(sumPrice) FROM KibanaSampleDataEcommerce WHERE 1 = 0".to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert_eq!(
        logical_plan.find_cube_scan().request,
        V1LoadRequestQuery {
            measures: Some(vec![]),
            segments: Some(vec![]),
            dimensions: Some(vec![]),
            order: Some(vec![]),
            limit: None,
            ungrouped: Some(true),
            ..Default::default()
        }
    );

    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;

    // Final query uses grouped query to Cube.js with WHERE FALSE, but without LIMIT 0
    assert!(!sql.contains("\"ungrouped\":"));
    assert!(sql.contains(r#"\"sql\":\"FALSE\""#));
    assert!(sql.contains(r#""limit": 50000"#));
}

/// Aggregation(dimension) with falsy filter should NOT get pushed to CubeScan with limit=0
/// This test currently produces WrappedSelect with WHERE FALSE, which is OK for our purposes
#[tokio::test]
async fn wrapper_dimension_agg_where_false() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            MAX(customer_gender)
        FROM
            KibanaSampleDataEcommerce
        WHERE 1 = 0
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let logical_plan = query_plan.as_logical_plan();
    assert_eq!(
        logical_plan.find_cube_scan().request,
        V1LoadRequestQuery {
            measures: Some(vec![]),
            dimensions: Some(vec![]),
            segments: Some(vec![]),
            order: Some(vec![]),
            limit: None,
            ungrouped: Some(true),
            ..Default::default()
        }
    );

    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;

    // Final query uses grouped query to Cube.js with WHERE FALSE, but without LIMIT 0
    assert!(!sql.contains("\"ungrouped\":"));
    assert!(sql.contains(r#"\"sql\":\"FALSE\""#));
    assert!(!sql.contains(r#""limit""#));
    assert!(sql.contains("LIMIT 50000"));
}

/// MIN(avg_measure) should get pushed to Cube with replaced measure
#[tokio::test]
async fn wrapper_min_from_avg_measure() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            MIN(avgPrice)
        FROM
            MultiTypeCube
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert_eq!(
        query_plan
            .as_logical_plan()
            .find_cube_scan_wrapped_sql()
            .request,
        TransportLoadRequestQuery {
            measures: Some(vec![json!({
                "cubeName": "MultiTypeCube",
                "alias": "min_multitypecub",
                "expr": {
                    "type": "PatchMeasure",
                    "sourceMeasure": "MultiTypeCube.avgPrice",
                    "replaceAggregationType": "min",
                    "addFilters": [],
                },
                "groupingSet": null,
            })
            .to_string(),]),
            dimensions: Some(vec![]),
            segments: Some(vec![]),
            order: Some(vec![]),
            ..Default::default()
        }
    );
}

#[tokio::test]
async fn test_ad_hoc_measure_filter() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"SELECT
    dim_str0,
    AVG(
        CASE (
            (
                CAST(TRUNC(EXTRACT(YEAR FROM dim_date0)) AS INTEGER) = 2024
            )
            AND
            (
                CAST(TRUNC(EXTRACT(MONTH FROM dim_date0)) AS INTEGER) <= 11
            )
        )
        WHEN TRUE
        THEN avgPrice
        ELSE NULL
        END
    ),
    SUM(
        CASE (dim_str1 = 'foo')
        WHEN TRUE
        THEN maxPrice
        ELSE NULL
        END
    )
FROM MultiTypeCube
GROUP BY
    1
;"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert_eq!(
        query_plan
            .as_logical_plan()
            .find_cube_scan_wrapped_sql()
            .request,
        TransportLoadRequestQuery {
            measures: Some(vec![
                json!({
                    "cubeName": "MultiTypeCube",
                    "alias": "avg_case_cast_tr",
                    "expr": {
                        "type": "PatchMeasure",
                        "sourceMeasure": "MultiTypeCube.avgPrice",
                        "replaceAggregationType": null,
                        "addFilters": [{
                            "cubeParams": ["MultiTypeCube"],
                            "sql": "(((CAST(TRUNC(EXTRACT(year FROM ${MultiTypeCube.dim_date0})) AS INTEGER) = 2024) AND (CAST(TRUNC(EXTRACT(month FROM ${MultiTypeCube.dim_date0})) AS INTEGER) <= 11)) = TRUE)"
                        }],
                    },
                    "groupingSet": null,
                }).to_string(),
                json!({
                    "cubeName": "MultiTypeCube",
                    "alias": "sum_case_multity",
                    "expr": {
                        "type": "PatchMeasure",
                        "sourceMeasure": "MultiTypeCube.maxPrice",
                        "replaceAggregationType": "sum",
                        "addFilters": [{
                            "cubeParams": ["MultiTypeCube"],
                            "sql": "((${MultiTypeCube.dim_str1} = $0$) = TRUE)"
                        }],
                    },
                    "groupingSet": null,
                }).to_string(),
            ]),
            dimensions: Some(vec![json!({
                "cubeName": "MultiTypeCube",
                "alias": "dim_str0",
                "expr": {
                    "type": "SqlFunction",
                    "cubeParams": ["MultiTypeCube"],
                    "sql": "${MultiTypeCube.dim_str0}",
                },
                "groupingSet": null,
            }).to_string(),]),
            segments: Some(vec![]),
            order: Some(vec![]),
            ..Default::default()
        }
    );
}

/// Backslashes in patch measure filter SQL must be escaped for JS template literal
/// evaluation on the Cube side, same as SqlFunction member expressions
#[tokio::test]
async fn test_ad_hoc_measure_filter_like_escape() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan_customized(
        // language=PostgreSQL
        r#"SELECT
    SUM(
        CASE (dim_str0 ILIKE '%foo%')
        WHEN TRUE
        THEN maxPrice
        ELSE NULL
        END
    )
FROM MultiTypeCube
;"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
        vec![(
            "expressions/ilike".to_string(),
            "{{ expr }} {% if negated %}NOT {% endif %}ILIKE {{ pattern }}\
             {% if default_escape %} ESCAPE '\\\\'{% endif %}"
                .to_string(),
        )],
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let request = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .request;
    let measures = request.measures.unwrap();
    assert_eq!(measures.len(), 1);
    let measure: serde_json::Value = serde_json::from_str(&measures[0]).unwrap();
    let filter_sql = measure["expr"]["addFilters"][0]["sql"].as_str().unwrap();
    // Member expression SQL is evaluated as a JS template literal on the Cube side,
    // which collapses `\\` back to `\`, so the ESCAPE clause must carry doubled backslashes
    assert!(
        filter_sql.contains(r#"ILIKE $0$ ESCAPE '\\\\'"#),
        "expected escaped ESCAPE clause in patch measure filter, got: {}",
        filter_sql
    );
}

#[tokio::test]
async fn test_wrapper_between() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            customer_gender
        FROM KibanaSampleDataEcommerce
        WHERE
            KibanaSampleDataEcommerce.customer_gender = customer_gender
            AND order_date BETWEEN '2024-01-01' AND '2024-12-31'
        GROUP BY 1
        ;"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    assert!(query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql
        .contains("BETWEEN $1 AND $2"));
}

#[tokio::test]
async fn test_wrapper_between_timestamp_date_bounds() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    // `order_date` is a TIMESTAMP time dimension; the BETWEEN bounds are DATE-typed
    // (`CURRENT_DATE` / `CURRENT_DATE - INTERVAL '52 weeks'`). PlanNormalize coerces the bounds
    // to the tested TIMESTAMP so all three operands share one type; here they fold to TIMESTAMP
    // literals. Without this, strict engines (e.g. BigQuery) reject the mixed-type BETWEEN.
    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
        SELECT
            customer_gender
        FROM KibanaSampleDataEcommerce
        WHERE
            KibanaSampleDataEcommerce.customer_gender = customer_gender
            AND order_date BETWEEN CURRENT_DATE - INTERVAL '52 weeks' AND CURRENT_DATE
        GROUP BY 1
        ;"#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let sql = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql
        .sql;
    println!("Generated SQL: {sql}");

    // Both bounds must end up as the same TIMESTAMP type as `order_date`: `CURRENT_DATE` and
    // `CURRENT_DATE - INTERVAL '52 weeks'` are coerced and fold to TIMESTAMP literals. Assert the
    // structural shape (rather than concrete dates) to stay independent of the wall clock.
    let between_re = Regex::new(
        r"\$\{KibanaSampleDataEcommerce\.order_date\} BETWEEN timestamptz '[^']+' AND timestamptz '[^']+'",
    )
    .unwrap();
    assert!(
        between_re.is_match(&sql),
        "expected both BETWEEN bounds coerced to TIMESTAMP literals, got: {}",
        sql
    );
    // No DATE / INTERVAL operands should leak through to the generated SQL.
    let upper = sql.to_uppercase();
    assert!(
        !upper.contains("CURRENT_DATE") && !upper.contains("INTERVAL"),
        "DATE/INTERVAL bound should have been coerced away, got: {}",
        sql
    );
}

#[tokio::test]
async fn test_wrapper_subqueries_filter_params_not_mixed() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        // language=PostgreSQL
        r#"
SELECT
    customer_gender
FROM KibanaSampleDataEcommerce
WHERE
    customer_gender IN (
        SELECT
            customer_gender
        FROM KibanaSampleDataEcommerce
        WHERE LOWER(notes) = 'sub_notes'
        GROUP BY 1
    )
    AND notes IN (
        SELECT
            notes
        FROM KibanaSampleDataEcommerce
        WHERE LOWER(notes) IN ('male_notes', 'female_notes', 'other_notes')
        GROUP BY 1
    )
GROUP BY 1
;
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );

    let wrapped_sql = query_plan
        .as_logical_plan()
        .find_cube_scan_wrapped_sql()
        .wrapped_sql;

    // Each subquery is generated with its own params; on merge, placeholders must be
    // remapped to the combined values so params don't mix between subqueries
    let placeholder_re = Regex::new(r"\$(\d+)\b").unwrap();
    let params_in_sql_order = placeholder_re
        .captures_iter(&wrapped_sql.sql)
        .map(|c| {
            let index = c[1].parse::<usize>().unwrap() - 1;
            wrapped_sql.values[index].clone().unwrap()
        })
        .collect::<Vec<_>>();
    assert_eq!(
        params_in_sql_order,
        vec!["sub_notes", "male_notes", "female_notes", "other_notes"]
    );
}

/// Date-only string literals (e.g. `'2026-06-24'`) compared to a timestamp column
/// inside an aggregated CASE expression must be normalized to timestamps
/// so the whole query can be pushed down to the wrapper.
#[tokio::test]
async fn test_case_wrapper_sum_case_date_only_string() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        r#"
        SELECT
            customer_gender,
            SUM(CASE WHEN order_date >= '2026-06-24' AND order_date < '2026-06-25' THEN sumPrice END) AS today_amount,
            MEASURE(sumPrice) AS mtd_amount
        FROM KibanaSampleDataEcommerce
        WHERE customer_gender = 'female'
            AND order_date >= '2026-06-01'
            AND order_date <= '2026-06-24'
        GROUP BY 1
        HAVING MEASURE(sumPrice) != 0
        ORDER BY 3 DESC
        LIMIT 100
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("2026-06-24T00:00:00.000Z"),
        "expected date-only string normalized to timestamp, got: {}",
        sql
    );

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

/// Query can reference no members at all, only synthetic fields.
/// Data source is resolved from cubes of the scan node in that case.
#[tokio::test]
async fn test_wrapper_only_system_fields() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        r#"
        SELECT COUNT(DISTINCT "F0"."__user") AS "user_count",
               COUNT(1) AS "row_count",
               MIN("F0"."__user") AS "user_min",
               MAX("F0"."__user") AS "user_max",
               COUNT(DISTINCT "F0"."__cubeJoinField") AS "join_field_count",
               MIN("F0"."__cubeJoinField") AS "join_field_min",
               MAX("F0"."__cubeJoinField") AS "join_field_max"
        FROM (
            SELECT "T1"."__user" AS "__user", "T1"."__cubeJoinField" AS "__cubeJoinField"
            FROM KibanaSampleDataEcommerce AS "T1"
        ) AS "F0"
        LIMIT 1
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains(r#"\"cubeName\":\"KibanaSampleDataEcommerce\""#),
        "SQL contains member expressions for the scanned cube: {}",
        sql
    );

    let physical_plan = query_plan.as_physical_plan().await.unwrap();
    println!(
        "Physical plan: {}",
        displayable(physical_plan.as_ref()).indent()
    );
}

/// A per-group aggregate in a CTE, date-filtered and counted on the outside, must be
/// pushed down whole: `DATEADD` in the outer filter is rewritten to `DATE_ADD`, so the
/// filter and the aggregate above it only push down when the data source has a
/// `functions/DATE_ADD` template.
#[tokio::test]
async fn test_wrapper_cte_aggregate_then_date_filter() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let query_plan = convert_select_to_query_plan(
        r#"
        WITH first_orders AS (
            SELECT customer_gender, MIN(order_date) AS first_order_at
            FROM KibanaSampleDataEcommerce
            WHERE has_subscription = true
            GROUP BY 1
        )
        SELECT COUNT(DISTINCT customer_gender) AS customers
        FROM first_orders
        WHERE first_order_at >= DATEADD('month', -12, CURRENT_DATE())
          AND first_order_at < CURRENT_DATE()
        "#
        .to_string(),
        DatabaseProtocol::PostgreSQL,
    )
    .await;

    let logical_plan = query_plan.as_logical_plan();
    let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
    assert!(
        sql.contains("COUNT(DISTINCT"),
        "outer aggregate is pushed down: {}",
        sql
    );
    assert!(
        sql.contains("DATE_ADD"),
        "outer date filter is pushed down: {}",
        sql
    );

    let _physical_plan = query_plan.as_physical_plan().await.unwrap();
}

/// `DATEADD` is rewritten to `date_add`, and the dialect template renders it from the
/// `date_part` and `interval` variables rather than from the arguments. The rewrite maps
/// every unit onto one of three parts - sub-day units become `MILLISECOND`, `day` and
/// `week` become `DAY`, and `month`, `quarter` and `year` become `MONTH` - so each dialect
/// has to render all three. These are the real templates from the query classes; the ones
/// used elsewhere in these tests take `args_concat` and would not catch a wrong unit.
#[tokio::test]
async fn test_wrapper_date_add_dialect_templates() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let dialects = [
        (
            // PostgresQuery, RedshiftQuery inherits it, DuckDBQuery repeats it
            "({{ args[0] }} + '{{ interval }} {{ date_part }}'::interval)",
            [
                "(CURRENT_DATE() + '7200000 MILLISECOND'::interval)",
                "(CURRENT_DATE() + '14 DAY'::interval)",
                "(CURRENT_DATE() + '24 MONTH'::interval)",
            ],
        ),
        (
            // SnowflakeQuery, MssqlQuery, RedshiftQuery
            "DATEADD({{ date_part }}, {{ interval }}, {{ args[0] }})",
            [
                "DATEADD(MILLISECOND, 7200000, CURRENT_DATE())",
                "DATEADD(DAY, 14, CURRENT_DATE())",
                "DATEADD(MONTH, 24, CURRENT_DATE())",
            ],
        ),
        (
            // MysqlQuery: MySQL has no MILLISECOND unit, so those become microseconds
            "DATE_ADD({{ args[0] }}, INTERVAL {% if date_part == \"MILLISECOND\" %}\
             {{ interval }}000 MICROSECOND{% else %}{{ interval }} {{ date_part }}{% endif %})",
            [
                "DATE_ADD(CURRENT_DATE(), INTERVAL 7200000000 MICROSECOND)",
                "DATE_ADD(CURRENT_DATE(), INTERVAL 14 DAY)",
                "DATE_ADD(CURRENT_DATE(), INTERVAL 24 MONTH)",
            ],
        ),
        (
            // ClickHouseQuery, DatabricksQuery
            "({{ args[0] }} + INTERVAL {{ interval }} {{ date_part }})",
            [
                "(CURRENT_DATE() + INTERVAL 7200000 MILLISECOND)",
                "(CURRENT_DATE() + INTERVAL 14 DAY)",
                "(CURRENT_DATE() + INTERVAL 24 MONTH)",
            ],
        ),
        (
            // PrestodbQuery, TrinoQuery and AthenaQuery inherit it
            "DATE_ADD('{{ date_part }}', {{ interval }}, {{ args[0] }})",
            [
                "DATE_ADD('MILLISECOND', 7200000, CURRENT_DATE())",
                "DATE_ADD('DAY', 14, CURRENT_DATE())",
                "DATE_ADD('MONTH', 24, CURRENT_DATE())",
            ],
        ),
    ];

    for (template, expected) in dialects {
        for (unit, expected) in ["hour", "week", "year"].iter().zip(expected) {
            // A filter over a per-group aggregate cannot become a Cube filter, so the
            // whole expression has to be rendered by the template
            let sql = convert_select_to_query_plan_customized(
                format!(
                    r#"
                    WITH first_orders AS (
                        SELECT customer_gender, MIN(order_date) AS first_order_at
                        FROM KibanaSampleDataEcommerce
                        GROUP BY 1
                    )
                    SELECT COUNT(DISTINCT customer_gender) AS customers
                    FROM first_orders
                    WHERE first_order_at > DATEADD('{unit}', 2, CURRENT_DATE())
                    "#
                ),
                DatabaseProtocol::PostgreSQL,
                vec![("functions/DATE_ADD".to_string(), template.to_string())],
            )
            .await
            .as_logical_plan()
            .find_cube_scan_wrapped_sql()
            .wrapped_sql
            .sql;

            assert!(
                sql.contains(expected),
                "`{}` renders as `{}` with template `{}`, got: {}",
                unit,
                expected,
                template,
                sql
            );
        }
    }
}

/// Sub-day units are reported as a millisecond count inside an `IntervalDayTime`, which
/// keeps days and milliseconds in separate halves of an i64 and reads them back
/// separately. A negative offset used to decode as a mixed interval and fail the query,
/// and a count past the 31 bit half used to wrap silently to a different amount of time,
/// so both signs and the range boundary are covered here. Whole days are carried into the
/// other half, which is what keeps ordinary spans past that boundary expressible. The
/// encoding is the same for every dialect, so one template stands in for all of them.
#[tokio::test]
async fn test_wrapper_date_add_negative_and_out_of_range_intervals() {
    if !Rewriter::sql_push_down_enabled() {
        return;
    }
    init_testing_logger();

    let context = TestContext::with_custom_templates(
        DatabaseProtocol::PostgreSQL,
        vec![(
            "functions/DATE_ADD".to_string(),
            "DATEADD({{ date_part }}, {{ interval }}, {{ args[0] }})".to_string(),
        )],
    )
    .await;

    let cases = [
        ("hour", "-2", Some("DATEADD(MILLISECOND, -7200000, ")),
        ("minute", "-30", Some("DATEADD(MILLISECOND, -1800000, ")),
        ("second", "-90", Some("DATEADD(MILLISECOND, -90000, ")),
        ("day", "-2", Some("DATEADD(DAY, -2, ")),
        ("week", "-2", Some("DATEADD(DAY, -14, ")),
        ("month", "-12", Some("DATEADD(MONTH, -12, ")),
        ("year", "-1", Some("DATEADD(MONTH, -12, ")),
        // The last span the millisecond half can hold on its own, and the first one past
        // it, which is day aligned and so is carried into the other half
        ("hour", "596", Some("DATEADD(MILLISECOND, 2145600000, ")),
        ("hour", "720", Some("DATEADD(DAY, 30, ")),
        ("hour", "-720", Some("DATEADD(DAY, -30, ")),
        ("minute", "43200", Some("DATEADD(DAY, 30, ")),
        // Neither day aligned nor small enough to count in milliseconds. The rewrite is
        // skipped, which leaves the `dateadd` stub to report `NotImplemented` at
        // execution rather than answering with a different date
        ("hour", "597", None),
        ("hour", "1000000", None),
    ];

    for (unit, amount, expected) in cases {
        let logical_plan = context
            .convert_sql_to_cube_query(&format!(
                r#"
                WITH first_orders AS (
                    SELECT customer_gender, MIN(order_date) AS first_order_at
                    FROM KibanaSampleDataEcommerce
                    GROUP BY 1
                )
                SELECT COUNT(DISTINCT customer_gender) AS customers
                FROM first_orders
                WHERE first_order_at > DATEADD('{unit}', {amount}, CURRENT_DATE())
                "#
            ))
            .await
            .unwrap_or_else(|error| panic!("`{}` by {} should compile: {}", unit, amount, error))
            .as_logical_plan();

        let Some(expected) = expected else {
            assert!(
                logical_plan.find_filter().is_some(),
                "`{}` by {} is not rewritten: {:?}",
                unit,
                amount,
                logical_plan
            );
            continue;
        };

        let sql = logical_plan.find_cube_scan_wrapped_sql().wrapped_sql.sql;
        assert!(
            sql.contains(expected),
            "`{}` by {} renders as `{}`, got: {}",
            unit,
            amount,
            expected,
            sql
        );
    }
}
