Repository: risingwavelabs/risingwave (language: python)

Issue:
shared source doesn't work with joins
### Describe the bug

nexmark q3

We can see `_rw_kafka_partition` is the cause.
```
Caused by:
    statement failed: db error: ERROR: Panicked when handling the request: a stream key is expected but not exist, plan:
    StreamHashJoin [append_only] { type: Inner, predicate: seller = _rw_kafka_partition }
    ├─StreamExchange { dist: HashShard(seller) }
    │ └─StreamFilter { predicate: (category = 10:Int32) }
    │   └─StreamRowIdGen { row_id_index: 11 }
    │     └─StreamSourceScan { columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset] }
    └─StreamExchange { dist: HashShard(id) }
      └─StreamFilter { predicate: (((state = 'or':Varchar) OR (state = 'id':Varchar)) OR (state = 'ca':Varchar)) }
        └─StreamRowIdGen { row_id_index: 9 }
          └─StreamSourceScan { columns: [id, name, email_address, credit_card, city, state, date_time, extra, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset] }
    This is a bug. We would appreciate a bug report at:
      https://github.com/risingwavelabs/risingwave/issues/new?labels=type%2Fbug&template=bug_report.yml
```

### Error message/log

_No response_

### To Reproduce

_No response_

### Expected behavior

_No response_

### How did you deploy RisingWave?

_No response_

### The version of RisingWave

_No response_

### Additional context

_No response_
fix(meta): locate upstream source fragment correctly for SourceBackfill
I hereby agree to the terms of the [RisingWave Labs, Inc. Contributor License Agreement](https://gist.github.com/TennyZhuang/f00be7f16996ea48effb049aa7be4d66#file-rw_cla).

## What's changed and what's your intention?
fix #19563

Previously, we use `upstream_fragment_ids` and `upstream_actor_ids` in the `Fragment` and `Actor` models directly to find the upstream Source of the SourceBackfill. 

The problem is that we assumed the fragment for SourceBackfill have only one upstream fragment, which is false. There could be other Merge nodes in the fragment, thus other upstream fragments. DynamicFilter is one example where the assumption doesn't hold.

In this PR, we change to use `upstream_fragment_id` in the Merge node.
Regarding `upstream_actor_id`, it's a little troublesome. Seem code and comments for more details.
 

<!--

**Please do not leave this empty!**

Please explain **IN DETAIL** what the changes are in this PR and why they are needed:

- Summarize your change (**mandatory**)
- How does this PR work? Need a brief introduction for the changed logic (optional)
- Describe clearly one logical change and avoid lazy messages (optional)
- Describe any limitations of the current code (optional)
- Refer to a related PR or issue link (optional)

-->

## Checklist

- [ ] I have written necessary rustdoc comments
- [ ] I have added necessary unit tests and integration tests
- [ ] I have added test labels as necessary. See [details](https://github.com/risingwavelabs/risingwave/blob/main/docs/developer-guide.md#ci-labels-guide).
- [ ] I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features #7934).
- [ ] My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
- [ ] All checks passed in `./risedev check` (or alias, `./risedev c`)
- [ ] My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)
<!-- To manually trigger a benchmark, please check out [Notion](https://www.notion.so/risingwave-labs/Manually-trigger-nexmark-performance-dashboard-test-b784f1eae1cf48889b2645d020b6b7d3). -->
- [ ] My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the [details](https://github.com/risingwavelabs/risingwave/blob/main/CONTRIBUTING.md))

## Documentation

- [ ] My PR needs documentation updates. (Please use the **Release note** section below to summarize the impact on users)

## Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.


<!--
Please create a release note for your changes.

Discuss technical details in the "What's changed" section, and
focus on the impact on users in the release note.

You should also mention the environment or conditions where the impact may occur.
-->

</details>

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:
Similar to https://github.com/risingwavelabs/risingwave/issues/17880. Hidden column isn't handled with care.



Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index cb54c1606356e..78574872d8f01 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -6,6 +6,7 @@ repos:
     rev: v2.3.0
     hooks:
     -   id: end-of-file-fixer
+        exclude: 'src/frontend/planner_test/tests/testdata/.*'
     -   id: trailing-whitespace
 -   repo: https://github.com/crate-ci/typos
     rev: v1.23.1
diff --git a/src/connector/src/parser/additional_columns.rs b/src/connector/src/parser/additional_columns.rs
index e5712d95066cb..62febde16243a 100644
--- a/src/connector/src/parser/additional_columns.rs
+++ b/src/connector/src/parser/additional_columns.rs
@@ -281,9 +281,20 @@ pub fn build_additional_column_desc(
 pub fn source_add_partition_offset_cols(
     columns: &[ColumnCatalog],
     connector_name: &str,
+    skip_col_id: bool,
 ) -> ([bool; 2], [ColumnDesc; 2]) {
     let mut columns_exist = [false; 2];
+
     let mut last_column_id = max_column_id(columns);
+    let mut assign_col_id = || {
+        if skip_col_id {
+            // col id will be filled outside later. Here just use a placeholder.
+            ColumnId::placeholder()
+        } else {
+            last_column_id = last_column_id.next();
+            last_column_id
+        }
+    };
 
     let additional_columns: Vec<_> = {
         let compat_col_types = COMPATIBLE_ADDITIONAL_COLUMNS
@@ -292,11 +303,10 @@ pub fn source_add_partition_offset_cols(
         ["partition", "file", "offset"]
             .iter()
             .filter_map(|col_type| {
-                last_column_id = last_column_id.next();
                 if compat_col_types.contains(col_type) {
                     Some(
                         build_additional_column_desc(
-                            last_column_id,
+                            assign_col_id(),
                             connector_name,
                             col_type,
                             None,
diff --git a/src/connector/src/source/reader/desc.rs b/src/connector/src/source/reader/desc.rs
index af607d2537ea6..d64435ef3a3cc 100644
--- a/src/connector/src/source/reader/desc.rs
+++ b/src/connector/src/source/reader/desc.rs
@@ -91,7 +91,7 @@ impl SourceDescBuilder {
             .map(|s| s.to_lowercase())
             .unwrap();
         let (columns_exist, additional_columns) =
-            source_add_partition_offset_cols(&self.columns, &connector_name);
+            source_add_partition_offset_cols(&self.columns, &connector_name, false);
 
         let mut columns: Vec<_> = self
             .columns
diff --git a/src/frontend/src/handler/create_source.rs b/src/frontend/src/handler/create_source.rs
index 1f89494f64ce6..d9f4fcce5ff85 100644
--- a/src/frontend/src/handler/create_source.rs
+++ b/src/frontend/src/handler/create_source.rs
@@ -31,8 +31,10 @@ use risingwave_common::catalog::{
 use risingwave_common::license::Feature;
 use risingwave_common::secret::LocalSecretManager;
 use risingwave_common::types::DataType;
+use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_connector::parser::additional_columns::{
     build_additional_column_desc, get_supported_additional_columns,
+    source_add_partition_offset_cols,
 };
 use risingwave_connector::parser::{
     fetch_json_schema_and_map_to_columns, AvroParserConfig, DebeziumAvroParserConfig,
@@ -1493,6 +1495,7 @@ pub async fn bind_create_source_or_table_with_connector(
     col_id_gen: &mut ColumnIdGenerator,
     // `true` for "create source", `false` for "create table with connector"
     is_create_source: bool,
+    is_shared_non_cdc: bool,
     source_rate_limit: Option<u32>,
 ) -> Result<(SourceCatalog, DatabaseId, SchemaId)> {
     let session = &handler_args.session;
@@ -1554,6 +1557,21 @@ pub async fn bind_create_source_or_table_with_connector(
     if is_create_source {
         // must behind `handle_addition_columns`
         check_and_add_timestamp_column(&with_properties, &mut columns);
+
+        // For shared sources, we will include partition and offset cols in the SourceExecutor's *output*, to be used by the SourceBackfillExecutor.
+        // For shared CDC source, the schema is different. See debezium_cdc_source_schema, CDC_BACKFILL_TABLE_ADDITIONAL_COLUMNS
+        if is_shared_non_cdc {
+            let (columns_exist, additional_columns) = source_add_partition_offset_cols(
+                &columns,
+                &with_properties.get_connector().unwrap(),
+                true, // col_id filled below at col_id_gen.generate
+            );
+            for (existed, c) in columns_exist.into_iter().zip_eq_fast(additional_columns) {
+                if !existed {
+                    columns.push(ColumnCatalog::hidden(c));
+                }
+            }
+        }
     }
 
     // resolve privatelink connection for Kafka
@@ -1670,14 +1688,14 @@ pub async fn handle_create_source(
     let with_properties = bind_connector_props(&handler_args, &format_encode, true)?;
 
     let create_cdc_source_job = with_properties.is_shareable_cdc_connector();
-    let is_shared = create_cdc_source_job
-        || (with_properties.is_shareable_non_cdc_connector()
-            && session
-                .env()
-                .streaming_config()
-                .developer
-                .enable_shared_source
-            && session.config().streaming_use_shared_source());
+    let is_shared_non_cdc = with_properties.is_shareable_non_cdc_connector()
+        && session
+            .env()
+            .streaming_config()
+            .developer
+            .enable_shared_source
+        && session.config().streaming_use_shared_source();
+    let is_shared = create_cdc_source_job || is_shared_non_cdc;
 
     let (columns_from_resolve_source, mut source_info) = if create_cdc_source_job {
         bind_columns_from_source_for_cdc(&session, &format_encode)?
@@ -1705,6 +1723,7 @@ pub async fn handle_create_source(
         stmt.include_column_options,
         &mut col_id_gen,
         true,
+        is_shared_non_cdc,
         overwrite_options.source_rate_limit,
     )
     .await?;
@@ -1777,8 +1796,7 @@ pub mod tests {
     use std::sync::Arc;
 
     use risingwave_common::catalog::{
-        CDC_SOURCE_COLUMN_NUM, DEFAULT_DATABASE_NAME, DEFAULT_SCHEMA_NAME, OFFSET_COLUMN_NAME,
-        ROWID_PREFIX, TABLE_NAME_COLUMN_NAME,
+        CDC_SOURCE_COLUMN_NUM, DEFAULT_DATABASE_NAME, DEFAULT_SCHEMA_NAME, ROWID_PREFIX,
     };
     use risingwave_common::types::{DataType, StructType};
 
@@ -1932,15 +1950,29 @@ pub mod tests {
             .columns
             .iter()
             .map(|col| (col.name(), col.data_type().clone()))
-            .collect::<HashMap<&str, DataType>>();
+            .collect::<Vec<(&str, DataType)>>();
 
-        let expected_columns = maplit::hashmap! {
-            ROWID_PREFIX => DataType::Serial,
-            "payload" => DataType::Jsonb,
-            OFFSET_COLUMN_NAME => DataType::Varchar,
-            TABLE_NAME_COLUMN_NAME => DataType::Varchar,
-        };
-        assert_eq!(columns, expected_columns);
+        expect_test::expect![[r#"
+            [
+                (
+                    "payload",
+                    Jsonb,
+                ),
+                (
+                    "_rw_offset",
+                    Varchar,
+                ),
+                (
+                    "_rw_table_name",
+                    Varchar,
+                ),
+                (
+                    "_row_id",
+                    Serial,
+                ),
+            ]
+        "#]]
+        .assert_debug_eq(&columns);
     }
 
     #[tokio::test]
@@ -1969,16 +2001,41 @@ pub mod tests {
             .unwrap();
         assert_eq!(source.name, "s");
 
-        let columns = GET_COLUMN_FROM_CATALOG(source);
-        let expect_columns = maplit::hashmap! {
-            ROWID_PREFIX => DataType::Serial,
-            "v1" => DataType::Int32,
-            "_rw_kafka_key" => DataType::Bytea,
-            // todo: kafka connector will automatically derive the column
-            // will change to a required field in the include clause
-            "_rw_kafka_timestamp" => DataType::Timestamptz,
-        };
-        assert_eq!(columns, expect_columns);
+        let columns = source
+            .columns
+            .iter()
+            .map(|col| (col.name(), col.data_type().clone()))
+            .collect::<Vec<(&str, DataType)>>();
+
+        expect_test::expect![[r#"
+            [
+                (
+                    "v1",
+                    Int32,
+                ),
+                (
+                    "_rw_kafka_key",
+                    Bytea,
+                ),
+                (
+                    "_rw_kafka_timestamp",
+                    Timestamptz,
+                ),
+                (
+                    "_rw_kafka_partition",
+                    Varchar,
+                ),
+                (
+                    "_rw_kafka_offset",
+                    Varchar,
+                ),
+                (
+                    "_row_id",
+                    Serial,
+                ),
+            ]
+        "#]]
+        .assert_debug_eq(&columns);
 
         let sql =
             "CREATE SOURCE s3 (v1 int) include timestamp 'header1' as header_col with (connector = 'kafka') format plain encode json"
diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
index a34c1cf462e0f..961be381b5f8c 100644
--- a/src/frontend/src/handler/create_table.rs
+++ b/src/frontend/src/handler/create_table.rs
@@ -506,6 +506,7 @@ pub(crate) async fn gen_create_table_plan_with_source(
         include_column_options,
         &mut col_id_gen,
         false,
+        false,
         rate_limit,
     )
     .await?;
diff --git a/src/frontend/src/handler/show.rs b/src/frontend/src/handler/show.rs
index cbd2effb54b8c..248636addee3a 100644
--- a/src/frontend/src/handler/show.rs
+++ b/src/frontend/src/handler/show.rs
@@ -684,7 +684,6 @@ pub fn handle_show_create_object(
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashMap;
     use std::ops::Index;
 
     use futures_async_stream::for_await;
@@ -720,36 +719,78 @@ mod tests {
         let sql = "show columns from t";
         let mut pg_response = frontend.run_sql(sql).await.unwrap();
 
-        let mut columns = HashMap::new();
+        let mut columns = Vec::new();
         #[for_await]
         for row_set in pg_response.values_stream() {
             let row_set = row_set.unwrap();
             for row in row_set {
-                columns.insert(
+                columns.push((
                     std::str::from_utf8(row.index(0).as_ref().unwrap())
                         .unwrap()
                         .to_string(),
                     std::str::from_utf8(row.index(1).as_ref().unwrap())
                         .unwrap()
                         .to_string(),
-                );
+                ));
             }
         }
 
-        let expected_columns: HashMap<String, String> = maplit::hashmap! {
-            "id".into() => "integer".into(),
-            "country.zipcode".into() => "character varying".into(),
-            "zipcode".into() => "bigint".into(),
-            "country.city.address".into() => "character varying".into(),
-            "country.address".into() => "character varying".into(),
-            "country.city".into() => "test.City".into(),
-            "country.city.zipcode".into() => "character varying".into(),
-            "rate".into() => "real".into(),
-            "country".into() => "test.Country".into(),
-            "_rw_kafka_timestamp".into() => "timestamp with time zone".into(),
-            "_row_id".into() => "serial".into(),
-        };
-
-        assert_eq!(columns, expected_columns);
+        expect_test::expect![[r#"
+            [
+                (
+                    "id",
+                    "integer",
+                ),
+                (
+                    "country",
+                    "test.Country",
+                ),
+                (
+                    "country.address",
+                    "character varying",
+                ),
+                (
+                    "country.city",
+                    "test.City",
+                ),
+                (
+                    "country.city.address",
+                    "character varying",
+                ),
+                (
+                    "country.city.zipcode",
+                    "character varying",
+                ),
+                (
+                    "country.zipcode",
+                    "character varying",
+                ),
+                (
+                    "zipcode",
+                    "bigint",
+                ),
+                (
+                    "rate",
+                    "real",
+                ),
+                (
+                    "_rw_kafka_timestamp",
+                    "timestamp with time zone",
+                ),
+                (
+                    "_rw_kafka_partition",
+                    "character varying",
+                ),
+                (
+                    "_rw_kafka_offset",
+                    "character varying",
+                ),
+                (
+                    "_row_id",
+                    "serial",
+                ),
+            ]
+        "#]]
+        .assert_debug_eq(&columns);
     }
 }
diff --git a/src/frontend/src/optimizer/optimizer_context.rs b/src/frontend/src/optimizer/optimizer_context.rs
index 75af6a3b3da69..2dfe766224ecb 100644
--- a/src/frontend/src/optimizer/optimizer_context.rs
+++ b/src/frontend/src/optimizer/optimizer_context.rs
@@ -191,7 +191,7 @@ impl OptimizerContext {
         }
         let mut optimizer_trace = self.optimizer_trace.borrow_mut();
         let string = str.into();
-        tracing::trace!(target: "explain_trace", "{}", string);
+        tracing::trace!(target: "explain_trace", "\n{}", string);
         optimizer_trace.push(string);
         optimizer_trace.push("\n".to_string());
     }
diff --git a/src/frontend/src/optimizer/plan_node/stream_source.rs b/src/frontend/src/optimizer/plan_node/stream_source.rs
index d7808e4be51ce..909fa1e0d3009 100644
--- a/src/frontend/src/optimizer/plan_node/stream_source.rs
+++ b/src/frontend/src/optimizer/plan_node/stream_source.rs
@@ -17,9 +17,6 @@ use std::rc::Rc;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pretty_xmlish::{Pretty, XmlNode};
-use risingwave_common::catalog::ColumnCatalog;
-use risingwave_common::util::iter_util::ZipEqFast;
-use risingwave_connector::parser::additional_columns::source_add_partition_offset_cols;
 use risingwave_pb::stream_plan::stream_node::PbNodeBody;
 use risingwave_pb::stream_plan::{PbStreamSource, SourceNode};
 
@@ -40,24 +37,7 @@ pub struct StreamSource {
 }
 
 impl StreamSource {
-    pub fn new(mut core: generic::Source) -> Self {
-        // For shared sources, we will include partition and offset cols in the *output*, to be used by the SourceBackfillExecutor.
-        // XXX: If we don't add here, these cols are also added in source reader, but pruned in the SourceExecutor's output.
-        // Should we simply add them here for all sources for consistency?
-        if let Some(source_catalog) = &core.catalog
-            && source_catalog.info.is_shared()
-        {
-            let (columns_exist, additional_columns) = source_add_partition_offset_cols(
-                &core.column_catalog,
-                &source_catalog.connector_name(),
-            );
-            for (existed, c) in columns_exist.into_iter().zip_eq_fast(additional_columns) {
-                if !existed {
-                    core.column_catalog.push(ColumnCatalog::hidden(c));
-                }
-            }
-        }
-
+    pub fn new(core: generic::Source) -> Self {
         let base = PlanBase::new_stream_with_core(
             &core,
             Distribution::SomeShard,
diff --git a/src/frontend/src/optimizer/plan_node/stream_source_scan.rs b/src/frontend/src/optimizer/plan_node/stream_source_scan.rs
index 83c79259952b2..ade1a4b6f2fee 100644
--- a/src/frontend/src/optimizer/plan_node/stream_source_scan.rs
+++ b/src/frontend/src/optimizer/plan_node/stream_source_scan.rs
@@ -17,11 +17,9 @@ use std::rc::Rc;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pretty_xmlish::{Pretty, XmlNode};
-use risingwave_common::catalog::{ColumnCatalog, Field};
+use risingwave_common::catalog::Field;
 use risingwave_common::types::DataType;
-use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_common::util::sort_util::OrderType;
-use risingwave_connector::parser::additional_columns::source_add_partition_offset_cols;
 use risingwave_pb::stream_plan::stream_node::{NodeBody, PbNodeBody};
 use risingwave_pb::stream_plan::PbStreamNode;
 
@@ -52,23 +50,7 @@ pub struct StreamSourceScan {
 impl_plan_tree_node_for_leaf! { StreamSourceScan }
 
 impl StreamSourceScan {
-    pub fn new(mut core: generic::Source) -> Self {
-        // XXX: do we need to include partition and offset cols here? It's needed by Backfill's input, but maybe not output?
-        // But the source's "schema" contains the hidden columns.
-        if let Some(source_catalog) = &core.catalog
-            && source_catalog.info.is_shared()
-        {
-            let (columns_exist, additional_columns) = source_add_partition_offset_cols(
-                &core.column_catalog,
-                &source_catalog.connector_name(),
-            );
-            for (existed, c) in columns_exist.into_iter().zip_eq_fast(additional_columns) {
-                if !existed {
-                    core.column_catalog.push(ColumnCatalog::hidden(c));
-                }
-            }
-        }
-
+    pub fn new(core: generic::Source) -> Self {
         let base = PlanBase::new_stream_with_core(
             &core,
             Distribution::SomeShard,


Test patch:
diff --git a/ci/scripts/e2e-source-test.sh b/ci/scripts/e2e-source-test.sh
index 4491db5633ea8..b6ea8267a1690 100755
--- a/ci/scripts/e2e-source-test.sh
+++ b/ci/scripts/e2e-source-test.sh
@@ -36,7 +36,7 @@ python3 -m pip install --break-system-packages requests protobuf fastavro conflu
 apt-get -y install jq
 
 echo "--- e2e, inline test"
-RUST_LOG="debug,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+RUST_LOG="debug,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info,risingwave_meta=info" \
 risedev ci-start ci-inline-source-test
 risedev slt './e2e_test/source_inline/**/*.slt' -j16
 risedev slt './e2e_test/source_inline/**/*.slt.serial'
diff --git a/e2e_test/source_inline/kafka/issue_19563.slt.serial b/e2e_test/source_inline/kafka/issue_19563.slt.serial
index 91b795d6acf34..528fac0735767 100644
--- a/e2e_test/source_inline/kafka/issue_19563.slt.serial
+++ b/e2e_test/source_inline/kafka/issue_19563.slt.serial
@@ -24,9 +24,9 @@ explain create materialized view mv1 as select v1 from kafkasource where v1 betw
 StreamMaterialize { columns: [v1, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
 └─StreamDynamicFilter { predicate: (v1 <= $expr1), output: [v1, _row_id], cleaned_by_watermark: true }
   ├─StreamProject { exprs: [v1, _row_id], output_watermarks: [v1] }
-  │ └─StreamDynamicFilter { predicate: (v1 >= now), output_watermarks: [v1], output: [v1, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset], cleaned_by_watermark: true }
-  │   ├─StreamRowIdGen { row_id_index: 2 }
-  │   │ └─StreamSourceScan { columns: [v1, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset] }
+  │ └─StreamDynamicFilter { predicate: (v1 >= now), output_watermarks: [v1], output: [v1, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id], cleaned_by_watermark: true }
+  │   ├─StreamRowIdGen { row_id_index: 4 }
+  │   │ └─StreamSourceScan { columns: [v1, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id] }
   │   └─StreamExchange { dist: Broadcast }
   │     └─StreamNow
   └─StreamExchange { dist: Broadcast }
diff --git a/src/frontend/planner_test/tests/testdata/output/batch_source.yaml b/src/frontend/planner_test/tests/testdata/output/batch_source.yaml
index 63daa9c43b2db..2b7f23dff932e 100644
--- a/src/frontend/planner_test/tests/testdata/output/batch_source.yaml
+++ b/src/frontend/planner_test/tests/testdata/output/batch_source.yaml
@@ -3,11 +3,11 @@
     select * from s
   logical_plan: |-
     LogicalProject { exprs: [id, value] }
-    └─LogicalSource { source: s, columns: [id, value, _rw_kafka_timestamp, _row_id] }
+    └─LogicalSource { source: s, columns: [id, value, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id] }
   batch_plan: |-
     BatchExchange { order: [], dist: Single }
     └─BatchProject { exprs: [id, value] }
-      └─BatchKafkaScan { source: s, columns: [id, value, _rw_kafka_timestamp, _row_id], filter: (None, None) }
+      └─BatchKafkaScan { source: s, columns: [id, value, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id], filter: (None, None) }
   create_source:
     format: plain
     encode: protobuf
diff --git a/src/frontend/planner_test/tests/testdata/output/shared_source.yml b/src/frontend/planner_test/tests/testdata/output/shared_source.yml
index 83fde26bfc7d2..fa75610ff09e3 100644
--- a/src/frontend/planner_test/tests/testdata/output/shared_source.yml
+++ b/src/frontend/planner_test/tests/testdata/output/shared_source.yml
@@ -68,12 +68,12 @@
   batch_plan: |-
     BatchExchange { order: [], dist: Single }
     └─BatchProject { exprs: [x, y] }
-      └─BatchKafkaScan { source: s, columns: [x, y, _rw_kafka_timestamp, _row_id], filter: (None, None) }
+      └─BatchKafkaScan { source: s, columns: [x, y, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id], filter: (None, None) }
   stream_plan: |-
     StreamMaterialize { columns: [x, y, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
     └─StreamProject { exprs: [x, y, _row_id] }
-      └─StreamRowIdGen { row_id_index: 3 }
-        └─StreamSourceScan { columns: [x, y, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset] }
+      └─StreamRowIdGen { row_id_index: 5 }
+        └─StreamSourceScan { columns: [x, y, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id] }
   with_config_map:
     streaming_use_shared_source: 'true'
 - before:
@@ -84,11 +84,11 @@
   batch_plan: |-
     BatchExchange { order: [], dist: Single }
     └─BatchProject { exprs: [x, y] }
-      └─BatchKafkaScan { source: s, columns: [x, y, _rw_kafka_timestamp, _row_id], filter: (None, None) }
+      └─BatchKafkaScan { source: s, columns: [x, y, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id], filter: (None, None) }
   stream_plan: |-
     StreamMaterialize { columns: [x, y, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
     └─StreamProject { exprs: [x, y, _row_id] }
-      └─StreamRowIdGen { row_id_index: 3 }
-        └─StreamSourceScan { columns: [x, y, _rw_kafka_timestamp, _row_id, _rw_kafka_partition, _rw_kafka_offset] }
+      └─StreamRowIdGen { row_id_index: 5 }
+        └─StreamSourceScan { columns: [x, y, _rw_kafka_timestamp, _rw_kafka_partition, _rw_kafka_offset, _row_id] }
   with_config_map:
     streaming_use_shared_source: 'true'