Repository: risingwavelabs/risingwave (language: python)

Issue:
Risingwave crashed for a highly complex insert sql query
### Describe the bug

I am supporting risingwave engine adapter for sqlmesh that is a data transformation tool. I am doing integration test and there exists a query that will crash risingwave and close the connection abnormally.

### Error message/log

```text
On the client side, the error is
"server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.
The connection to the server was lost. Attempting reset: Failed."

On the server side, the error is
"thread 'rw-batch' panicked at src/common/src/array/data_chunk.rs:186:10:
index out of bounds: the len is 4 but the index is 4
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread 'rw-standalone-compute' panicked at src/batch/src/executor/group_top_n.rs:119:41:
index out of bounds: the len is 4 but the index is 4"
```


### To Reproduce

```sql
CREATE SCHEMA IF NOT EXISTS "test_schema_h04rwzvk";

CREATE TABLE "test_schema_h04rwzvk"."test_table" (
    id INT,
    name TEXT,
    updated_at TIMESTAMP,
    valid_from TIMESTAMP,
    valid_to TIMESTAMP
);

CREATE TABLE "test_schema_h04rwzvk"."__temp_test_table_yjk0fdr8" (
    id INT,
    name TEXT,
    updated_at TIMESTAMP,
    valid_from TIMESTAMP,
    valid_to TIMESTAMP
);

INSERT INTO "test_schema_h04rwzvk"."__temp_test_table_yjk0fdr8" (id, name, updated_at, valid_from, valid_to)
VALUES
    (1, 'a', '2022-01-01 00:00:00', '2022-01-01 00:00:00', NULL),
    (2, 'b', '2022-01-02 00:00:00', '2022-01-02 00:00:00', NULL),
    (3, 'c', '2022-01-03 00:00:00', '2022-01-03 00:00:00', '2022-01-04 00:00:00');


INSERT INTO "test_schema_h04rwzvk"."test_table" 
    ("id", "name", "updated_at", "valid_from", "valid_to")
WITH 
    "source" AS (
        SELECT DISTINCT ON (COALESCE("id", -1)) 
            TRUE AS "_exists",
            "id",
            "name",
            CAST("updated_at" AS TIMESTAMP) AS "updated_at"
        FROM (
            SELECT 
                CAST("id" AS INT) AS "id",
                CAST("name" AS TEXT) AS "name",
                CAST("updated_at" AS TIMESTAMP) AS "updated_at"
            FROM (
                VALUES 
                    (1, 'a', CAST('2022-01-01 00:00:00' AS TIMESTAMP)),
                    (2, 'b', CAST('2022-01-02 00:00:00' AS TIMESTAMP)),
                    (3, 'c', CAST('2022-01-03 00:00:00' AS TIMESTAMP))
            ) AS "t"("id", "name", "updated_at")
        ) AS "raw_source"
    ),
    "static" AS (
        SELECT 
            "id", 
            "name", 
            "updated_at", 
            "valid_from", 
            "valid_to", 
            TRUE AS "_exists"
        FROM "dev"."test_schema_h04rwzvk"."__temp_test_table_yjk0fdr8"
        WHERE NOT "valid_to" IS NULL
        LIMIT 0
    ),
    "latest" AS (
        SELECT 
            "id", 
            "name", 
            "updated_at", 
            "valid_from", 
            "valid_to", 
            TRUE AS "_exists"
        FROM "dev"."test_schema_h04rwzvk"."__temp_test_table_yjk0fdr8"
        WHERE "valid_to" IS NULL
        LIMIT 0
    ),
    "deleted" AS (
        SELECT 
            "static"."id", 
            "static"."name", 
            "static"."updated_at", 
            "static"."valid_from", 
            "static"."valid_to"
        FROM "static"
        LEFT JOIN "latest" 
            ON COALESCE("static"."id", -1) = COALESCE("latest"."id", -1)
        WHERE "latest"."valid_to" IS NULL
    ),
    "latest_deleted" AS (
        SELECT 
            TRUE AS "_exists", 
            COALESCE("id", -1) AS "_key0", 
            MAX("valid_to") AS "valid_to"
        FROM "deleted"
        GROUP BY COALESCE("id", -1)
    ),
    "joined" AS (
        SELECT 
            "source"."_exists" AS "_exists",
            "latest"."id" AS "t_id",
            "latest"."name" AS "t_name",
            "latest"."updated_at" AS "t_updated_at",
            "latest"."valid_from" AS "t_valid_from",
            "latest"."valid_to" AS "t_valid_to",
            "source"."id" AS "id",
            "source"."name" AS "name",
            "source"."updated_at" AS "updated_at"
        FROM "latest"
        LEFT JOIN "source" 
            ON COALESCE("latest"."id", -1) = COALESCE("source"."id", -1)
        UNION ALL
        SELECT 
            "source"."_exists" AS "_exists",
            "latest"."id" AS "t_id",
            "latest"."name" AS "t_name",
            "latest"."updated_at" AS "t_updated_at",
            "latest"."valid_from" AS "t_valid_from",
            "latest"."valid_to" AS "t_valid_to",
            "source"."id" AS "id",
            "source"."name" AS "name",
            "source"."updated_at" AS "updated_at"
        FROM "latest"
        RIGHT JOIN "source" 
            ON COALESCE("latest"."id", -1) = COALESCE("source"."id", -1)
        WHERE "latest"."_exists" IS NULL
    ),
    "updated_rows" AS (
        SELECT 
            COALESCE("joined"."t_id", "joined"."id") AS "id",
            COALESCE("joined"."t_name", "joined"."name") AS "name",
            COALESCE("joined"."t_updated_at", "joined"."updated_at") AS "updated_at",
            CASE 
                WHEN "t_valid_from" IS NULL AND NOT "latest_deleted"."_exists" IS NULL 
                THEN CASE 
                    WHEN "latest_deleted"."valid_to" > "updated_at" 
                    THEN "latest_deleted"."valid_to" 
                    ELSE "updated_at" 
                END
                WHEN "t_valid_from" IS NULL 
                THEN CAST('1970-01-01 00:00:00' AS TIMESTAMP)
                ELSE "t_valid_from"
            END AS "valid_from",
            CASE 
                WHEN "joined"."updated_at" > "joined"."t_updated_at" 
                THEN "joined"."updated_at"
                WHEN "joined"."_exists" IS NULL 
                THEN CAST('2023-01-01 00:00:00' AS TIMESTAMP)
                ELSE "t_valid_to"
            END AS "valid_to"
        FROM "joined"
        LEFT JOIN "latest_deleted" 
            ON COALESCE("joined"."id", -1) = "latest_deleted"."_key0"
    ),
    "inserted_rows" AS (
        SELECT 
            "id", 
            "name", 
            "updated_at", 
            "updated_at" AS "valid_from", 
            CAST(NULL AS TIMESTAMP) AS "valid_to"
        FROM "joined"
        WHERE "joined"."updated_at" > "joined"."t_updated_at"
    )
SELECT 
    "id", 
    "name", 
    "updated_at", 
    "valid_from", 
    "valid_to"
FROM "static"
UNION ALL
SELECT 
    "id", 
    "name", 
    "updated_at", 
    "valid_from", 
    "valid_to"
FROM "updated_rows"
UNION ALL
SELECT 
    "id", 
    "name", 
    "updated_at", 
    "valid_from", 
    "valid_to"
FROM "inserted_rows";
```

### Expected behavior

The risingwave should not expect to crash even if the query is wrong or not successful.

### How did you deploy RisingWave?

I am using `risingwave playground` as test environment.

### The version of RisingWave

_No response_

### Additional context

_No response_
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:
<details>
<summary>stacktrace</summary>

```
   0: rust_begin_unwind
             at /rustc/52fd9983996d9fcfb719749838336be66dee68f9/library/std/src/panicking.rs:665:5
   1: core::panicking::panic_fmt
             at /rustc/52fd9983996d9fcfb719749838336be66dee68f9/library/core/src/panicking.rs:74:14
   2: core::panicking::panic_bounds_check
             at /rustc/52fd9983996d9fcfb719749838336be66dee68f9/library/core/src/panicking.rs:276:5
   3: <usize as core::slice::index::SliceIndex<[T]>>::index
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/slice/index.rs:301:10
   4: core::slice::index::<impl core::ops::index::Index<I> for [T]>::index
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/slice/index.rs:16:9
   5: <alloc::vec::Vec<T,A> as core::ops::index::Index<I>>::index
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:3323:9
   6: <risingwave_batch::executor::group_top_n::GroupTopNExecutorBuilder as risingwave_batch::executor::BoxedExecutorBuilder>::new_boxed_executor::{{closure}}::{{closure}}
             at ./src/batch/src/executor/group_top_n.rs:120:41
   7: core::iter::adapters::map::map_fold::{{closure}}
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:88:28
   8: <core::slice::iter::Iter<T> as core::iter::traits::iterator::Iterator>::fold
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/slice/iter/macros.rs:232:27
   9: <core::iter::adapters::map::Map<I,F> as core::iter::traits::iterator::Iterator>::fold
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/iter/adapters/map.rs:128:9
  10: core::iter::traits::iterator::Iterator::for_each
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:813:9
  11: alloc::vec::Vec<T,A>::extend_trusted
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:3524:17
  12: <alloc::vec::Vec<T,A> as alloc::vec::spec_extend::SpecExtend<T,I>>::spec_extend
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/spec_extend.rs:26:9
  13: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter_nested::SpecFromIterNested<T,I>>::from_iter
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/spec_from_iter_nested.rs:60:9
  14: <alloc::vec::Vec<T> as alloc::vec::spec_from_iter::SpecFromIter<T,I>>::from_iter
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/spec_from_iter.rs:33:9
  15: <alloc::vec::Vec<T> as core::iter::traits::collect::FromIterator<T>>::from_iter
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/vec/mod.rs:3388:9
  16: core::iter::traits::iterator::Iterator::collect
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/iter/traits/iterator.rs:2001:9
  17: <risingwave_batch::executor::group_top_n::GroupTopNExecutorBuilder as risingwave_batch::executor::BoxedExecutorBuilder>::new_boxed_executor::{{closure}}
             at ./src/batch/src/executor/group_top_n.rs:118:31
  18: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  19: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:258:10
  20: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  21: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  22: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:215:65
  23: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  24: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  25: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:215:65
  26: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  27: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  28: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:215:65
  29: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  30: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  31: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:215:65
  32: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  33: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  34: risingwave_batch::executor::ExecutorBuilder<C>::try_build::{{closure}}
             at ./src/batch/src/executor/mod.rs:215:65
  35: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  36: risingwave_batch::executor::ExecutorBuilder<C>::build::{{closure}}
             at ./src/batch/src/executor/mod.rs:202:14
  37: risingwave_frontend::scheduler::local::LocalQueryExecution::run_inner::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:122:41
  38: <futures_async_stream::try_stream::GenTryStream<G> as futures_core::stream::Stream>::poll_next
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-async-stream-0.2.11/src/lib.rs:492:33
  39: <tracing_futures::Instrumented<T> as futures_core::stream::Stream>::poll_next
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tracing-futures-0.2.5/src/lib.rs:342:9
  40: <core::pin::Pin<P> as futures_core::stream::Stream>::poll_next
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-core-0.3.30/src/stream.rs:120:9
  41: <futures_util::stream::stream::map::Map<St,F> as futures_core::stream::Stream>::poll_next
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.30/src/stream/stream/map.rs:58:26
  42: futures_util::stream::stream::StreamExt::poll_next_unpin
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.30/src/stream/stream/mod.rs:1638:9
  43: <futures_util::stream::stream::next::Next<St> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-util-0.3.30/src/stream/stream/next.rs:32:9
  44: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:157:56
  45: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  46: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  47: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  48: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:178:77
  49: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  50: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  51: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  52: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  53: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:179:81
  54: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  55: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  56: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  57: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  58: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:180:63
  59: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  60: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  61: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  62: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  63: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:181:71
  64: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  65: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  66: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  67: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  68: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:182:73
  69: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  70: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  71: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  72: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  73: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:183:67
  74: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  75: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  76: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  77: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  78: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:184:71
  79: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  80: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll::{{closure}}
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:391:31
  81: tokio::task::task_local::LocalKey<T>::scope_inner
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:217:19
  82: <tokio::task::task_local::TaskLocalFuture<T,F> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/task/task_local.rs:387:19
  83: risingwave_frontend::scheduler::local::LocalQueryExecution::stream_rows::{{closure}}
             at ./src/frontend/src/scheduler/local.rs:185:71
  84: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /Users/eric/.rustup/toolchains/nightly-2024-10-11-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9
  85: <tokio::time::timeout::Timeout<T> as core::future::future::Future>::poll
             at /Users/eric/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.38.0/src/time/timeout.rs:202:33
```

</details>



Fix patch:
diff --git a/src/frontend/src/optimizer/plan_node/generic/dedup.rs b/src/frontend/src/optimizer/plan_node/generic/dedup.rs
index 69749f4637fc3..3d93f7510afe4 100644
--- a/src/frontend/src/optimizer/plan_node/generic/dedup.rs
+++ b/src/frontend/src/optimizer/plan_node/generic/dedup.rs
@@ -28,6 +28,16 @@ pub struct Dedup<PlanRef> {
 }
 
 impl<PlanRef: GenericPlanRef> Dedup<PlanRef> {
+    pub fn new(input: PlanRef, dedup_cols: Vec<usize>) -> Self {
+        debug_assert!(
+            dedup_cols.iter().all(|&idx| idx < input.schema().len()),
+            "Invalid dedup keys {:?} input schema size = {}",
+            &dedup_cols,
+            input.schema().len()
+        );
+        Dedup { input, dedup_cols }
+    }
+
     fn dedup_cols_pretty<'a>(&self) -> Pretty<'a> {
         Pretty::Array(
             self.dedup_cols
diff --git a/src/frontend/src/optimizer/plan_node/generic/top_n.rs b/src/frontend/src/optimizer/plan_node/generic/top_n.rs
index 3fd19948e6c99..b8aac5a5ed062 100644
--- a/src/frontend/src/optimizer/plan_node/generic/top_n.rs
+++ b/src/frontend/src/optimizer/plan_node/generic/top_n.rs
@@ -113,6 +113,13 @@ impl<PlanRef: GenericPlanRef> TopN<PlanRef> {
             assert!(offset == 0, "WITH TIES is not supported with OFFSET");
         }
 
+        debug_assert!(
+            group_key.iter().all(|&idx| idx < input.schema().len()),
+            "Invalid group keys {:?} input schema size = {}",
+            &group_key,
+            input.schema().len()
+        );
+
         Self {
             input,
             limit_attr,
diff --git a/src/frontend/src/optimizer/plan_node/logical_dedup.rs b/src/frontend/src/optimizer/plan_node/logical_dedup.rs
index a9902bc87ac11..bd9393a45fef5 100644
--- a/src/frontend/src/optimizer/plan_node/logical_dedup.rs
+++ b/src/frontend/src/optimizer/plan_node/logical_dedup.rs
@@ -39,7 +39,7 @@ pub struct LogicalDedup {
 
 impl LogicalDedup {
     pub fn new(input: PlanRef, dedup_cols: Vec<usize>) -> Self {
-        let core = generic::Dedup { input, dedup_cols };
+        let core = generic::Dedup::new(input, dedup_cols);
         let base = PlanBase::new_logical_with_core(&core);
         LogicalDedup { base, core }
     }
@@ -163,7 +163,12 @@ impl ColPrunable for LogicalDedup {
         );
 
         let new_input = self.input().prune_col(&input_required_cols, ctx);
-        let logical_dedup = Self::new(new_input, self.dedup_cols().to_vec()).into();
+        let new_dedup_cols = self
+            .dedup_cols()
+            .iter()
+            .map(|&idx| mapping.map(idx))
+            .collect_vec();
+        let logical_dedup = Self::new(new_input, new_dedup_cols).into();
 
         if input_required_cols == required_cols {
             logical_dedup


Test patch:
diff --git a/src/frontend/planner_test/tests/testdata/input/distinct_on.yaml b/src/frontend/planner_test/tests/testdata/input/distinct_on.yaml
index 199314aba505f..1bbfc8ed97666 100644
--- a/src/frontend/planner_test/tests/testdata/input/distinct_on.yaml
+++ b/src/frontend/planner_test/tests/testdata/input/distinct_on.yaml
@@ -50,3 +50,9 @@
     select distinct on (4) * from t;
   expected_outputs:
     - binder_error
+# Issue #19608
+- sql: |
+    create table t (a int, b int, c int);
+    WITH s AS (SELECT DISTINCT ON (a+1) true as id, * FROM t) SELECT a,b,c from s;
+  expected_outputs:
+    - batch_plan
diff --git a/src/frontend/planner_test/tests/testdata/output/distinct_on.yaml b/src/frontend/planner_test/tests/testdata/output/distinct_on.yaml
index 54e0d58882ad3..061be65dd56bb 100644
--- a/src/frontend/planner_test/tests/testdata/output/distinct_on.yaml
+++ b/src/frontend/planner_test/tests/testdata/output/distinct_on.yaml
@@ -105,3 +105,13 @@
     create table t (a int, b int, c int);
     select distinct on (4) * from t;
   binder_error: 'Invalid input syntax: Invalid ordinal number in DISTINCT ON: 4'
+- sql: |
+    create table t (a int, b int, c int);
+    WITH s AS (SELECT DISTINCT ON (a+1) true as id, * FROM t) SELECT a,b,c from s;
+  batch_plan: |-
+    BatchExchange { order: [], dist: Single }
+    └─BatchProject { exprs: [t.a, t.b, t.c] }
+      └─BatchGroupTopN { order: [], limit: 1, offset: 0, group_key: [$expr1] }
+        └─BatchExchange { order: [], dist: HashShard($expr1) }
+          └─BatchProject { exprs: [t.a, t.b, t.c, (t.a + 1:Int32) as $expr1] }
+            └─BatchScan { table: t, columns: [t.a, t.b, t.c], distribution: SomeShard }