ESQL: Prevents mixing TS and STANDARD indexes with TS and views - #153366

FROM (TS k8s), (FROM employees)
| STATS bytes = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)
| SORT time_bucket

        if (plan instanceof UnionAll || plan instanceof Subquery) {
            return false;
        }

Q1
public void testTimeSeriesAggregateFunctionOutsideTsCommand() {
        analyzer().addK8s().error("""
            FROM k8s
            | STATS x = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)
            """, equalTo("""
            Found 1 problem
            line 2:13: time_series aggregate[last_over_time(event)] can only be used with the TS command"""));
}

Q2
public void testRateAggregateFunctionOutsideTsCommand() {
        analyzer().addK8s().error("""
            FROM k8s
            | STATS max_cost = max(rate(network.total_cost)) BY cluster
            """, equalTo("""
            Found 1 problem
            line 2:24: time_series aggregate[rate(network.total_cost)] can only be used with the TS command"""));
}

Q3
public void testTimeSeriesAggregateFunctionOutsideTsCommandAfterSubquery() {
        analyzer().addK8s().error("""
            FROM (FROM k8s), (FROM k8s)
            | STATS x = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)
            """, equalTo("""
            Found 1 problem
            line 2:13: time_series aggregate[last_over_time(event)] can only be used with the TS command"""));
}

    protected void checkTimeSeriesAggregates(Failures failures) {
        Holder<Boolean> isTimeSeriesIndexMode = new Holder<>(false);
        child().forEachDown(p -> { ===> it does not check UnionAll
            if (p instanceof EsRelation er && er.indexMode().isTsdb()) {
                isTimeSeriesIndexMode.set(true);
            }
        });
        if (isTimeSeriesIndexMode.get()) {
            return;
        }
        forEachExpression(
            TimeSeriesAggregateFunction.class,
            r -> failures.add(fail(r, "time_series aggregate[{}] can only be used with the TS command", r.sourceText()))
        );
    }

Q4
from (TS k8s), (FROM sample_data)
| STATS x = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)

    protected void checkTimeSeriesAggregates(Failures failures) {
        if (hasTimeSeriesSource(child())) {
            return;
        }
        Holder<Boolean> hasTimeSeriesAgg = new Holder<>(false);
        forEachExpression(TimeSeriesAggregateFunction.class, r -> hasTimeSeriesAgg.set(true));
        if (hasTimeSeriesAgg.get() == false) {
            return;
        }
        if (child().anyMatch(p -> p instanceof UnionAll)) {
            failures.add(
                fail(
                    this,
                    "time-series aggregation [{}] cannot be applied over a union of data sources; "
                        + "apply the time-series aggregation inside each subquery instead",
                    sourceText()
                )
            );
        } else {
            forEachExpression(
                TimeSeriesAggregateFunction.class,
                r -> failures.add(fail(r, "time_series aggregate[{}] can only be used with the TS command", r.sourceText()))
            );
        }
    }

    /**
     * Returns {@code true} if {@code plan} (or any non-{@link UnionAll} descendant, excluding the right-hand side
     * of an {@link AbstractSubqueryJoin}) holds an {@link EsRelation} in time-series index mode.
     * <p>
     * Traversal stops at {@link UnionAll} boundaries so that a {@code TS} source nested inside a {@code FROM}
     * subquery (e.g. {@code FROM (TS k8s), (FROM sample_data)}) does not allow time-series aggregate functions
     * in the outer {@code STATS}: the {@code TS} relation is isolated inside an independent subquery and does
     * not grant TS semantics to this aggregate.
     */
    private static boolean hasTimeSeriesSource(LogicalPlan plan) {
        if (plan instanceof EsRelation er && er.indexMode().isTsdb()) {
            return true;
        }
        if (plan instanceof UnionAll) {
            return false;
        }
        if (plan instanceof AbstractSubqueryJoin join) {
            // Only the left (main data) side can supply a time-series source; right is a lookup/subquery.
            return hasTimeSeriesSource(join.left());
        }
        for (LogicalPlan child : plan.children()) {
            if (hasTimeSeriesSource(child)) {
                return true;
            }
        }
        return false;
    }

    /**
     * Returns true if {@code plan} or any node reachable via the main data pipeline contains a {@link UnionAll}.
     * The right side of an {@link AbstractSubqueryJoin} (the independent subquery) is excluded because a
     * {@link UnionAll} there only affects the join lookup, not the time-series data stream.
     */
    private static boolean hasUnionAllInDataPath(LogicalPlan plan) {
        if (plan instanceof UnionAll) {
            return true;
        }
        if (plan instanceof AbstractSubqueryJoin join) {
            return hasUnionAllInDataPath(join.left());
        }
        for (LogicalPlan child : plan.children()) {
            if (hasUnionAllInDataPath(child)) {
                return true;
            }
        }
        return false;
    }

    public void verify(Failures failures) {
        if (hasUnionAllInDataPath(child())) {
            failures.add(
                fail(
                    this,
                    "time-series aggregation [{}] cannot be applied over a union of data sources; "
                        + "apply the time-series aggregation inside each subquery instead",
                    sourceText()
                )
            );
        }
    ...

FROM (TS k8s)
| STATS bytes = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)
| SORT time_bucket

TS k8s
| STATS bytes = last_over_time(event) BY time_bucket = bucket(@timestamp, 1 day)
| SORT time_bucket