ORCA: Support pushdown partial aggregate below join

[jiaqizho]

Fixes #ISSUE_Number

What does this PR do?

Type of Change

• Bug fix (non-breaking change)
• New feature (non-breaking change)
• Breaking change (fix or feature with breaking changes)
• Documentation update

Breaking Changes

Test Plan

• Unit tests added/updated
• Integration tests added/updated
• Passed make installcheck
• Passed make -C src/test installcheck-cbdb-parallel

Impact

Performance:

User-facing changes:

Dependencies:

Checklist

• Followed contribution guide
• Added/updated documentation
• Reviewed code for security implications
• Requested review from cloudberry committers

Additional Context

CI Skip Instructions

---

[my-ship-it]

Just tried:

-- 状态转换函数：处理单个输入值并更新状态
CREATE OR REPLACE FUNCTION avg_trans_fn(
    state numeric[],  -- 当前状态：[sum, count]
    next_val numeric  -- 新输入值
) RETURNS numeric[] AS $$
BEGIN
    IF next_val IS NOT NULL THEN
        state[1] := COALESCE(state[1], 0) + next_val;
        state[2] := COALESCE(state[2], 0) + 1;
    END IF;
    RETURN state;
END;
$$ LANGUAGE plpgsql;

-- 合并函数：合并两个部分聚合状态
CREATE OR REPLACE FUNCTION avg_combine_fn(
    state1 numeric[],
    state2 numeric[]
) RETURNS numeric[] AS $$
BEGIN
    IF state1 IS NULL THEN
        RETURN state2;
    ELSIF state2 IS NULL THEN
        RETURN state1;
    ELSE
        RETURN ARRAY[state1[1] + state2[1], state1[2] + state2[2]];
    END IF;
END;
$$ LANGUAGE plpgsql;

-- 最终函数：从状态计算最终结果
CREATE OR REPLACE FUNCTION avg_final_fn(
    state numeric[]
) RETURNS numeric AS $$
BEGIN
    IF state[2] = 0 THEN
        RETURN NULL;
    ELSE
        RETURN state[1] / state[2];
    END IF;
END;
$$ LANGUAGE plpgsql;

CREATE AGGREGATE my_avg(numeric) (
    sfunc = avg_trans_fn,       -- 状态转换函数
    stype = numeric[],          -- 状态类型
    combinefunc = avg_combine_fn, -- 合并函数
    finalfunc = avg_final_fn,   -- 最终计算函数
    initcond = '{0,0}'          -- 初始状态
);

CREATE TABLE tbl1(v1 int, v2 numeric);
CREATE TABLE tbl2(v1 int, v2 numeric);

The agg pushdown doesn't work in ORCA but works in postgres Postgres query optimizer.

[jiaqizho]

Just tried:

-- 状态转换函数：处理单个输入值并更新状态
CREATE OR REPLACE FUNCTION avg_trans_fn(
    state numeric[],  -- 当前状态：[sum, count]
    next_val numeric  -- 新输入值
) RETURNS numeric[] AS $$
BEGIN
    IF next_val IS NOT NULL THEN
        state[1] := COALESCE(state[1], 0) + next_val;
        state[2] := COALESCE(state[2], 0) + 1;
    END IF;
    RETURN state;
END;
$$ LANGUAGE plpgsql;

-- 合并函数：合并两个部分聚合状态
CREATE OR REPLACE FUNCTION avg_combine_fn(
    state1 numeric[],
    state2 numeric[]
) RETURNS numeric[] AS $$
BEGIN
    IF state1 IS NULL THEN
        RETURN state2;
    ELSIF state2 IS NULL THEN
        RETURN state1;
    ELSE
        RETURN ARRAY[state1[1] + state2[1], state1[2] + state2[2]];
    END IF;
END;
$$ LANGUAGE plpgsql;

-- 最终函数：从状态计算最终结果
CREATE OR REPLACE FUNCTION avg_final_fn(
    state numeric[]
) RETURNS numeric AS $$
BEGIN
    IF state[2] = 0 THEN
        RETURN NULL;
    ELSE
        RETURN state[1] / state[2];
    END IF;
END;
$$ LANGUAGE plpgsql;

CREATE AGGREGATE my_avg(numeric) (
    sfunc = avg_trans_fn,       -- 状态转换函数
    stype = numeric[],          -- 状态类型
    combinefunc = avg_combine_fn, -- 合并函数
    finalfunc = avg_final_fn,   -- 最终计算函数
    initcond = '{0,0}'          -- 初始状态
);

CREATE TABLE tbl1(v1 int, v2 numeric);
CREATE TABLE tbl2(v1 int, v2 numeric);

The agg pushdown doesn't work in ORCA but works in postgres Postgres query optimizer.

Without any data in the table tbl1 and tbl2. ORCA choose the one-stage agg, but we do have the pushdown path.

we can use the optimizer_force_multistage_agg to force ORCA choose the multi-stage path(But won't create the no-pushdown path, because the enforce of distribution).

also use the my_avg