bug: multiple reads in source's internal table

[tabVersion]

Describe the bug

write one row in state table and it read multiple identical rows when select * from the internal table

this issue is introduced in #5433

To Reproduce

1. start a cluster with kafka (./risedev d full & ./scripts/source/prepare_ci_kafka.sh)
2. create a materialized source create table s2 (v1 int, v2 varchar) with ( connector = 'kafka', topic = 'kafka_2_partition_topic', properties.bootstrap.server = '127.0.0.1:29092', scan.startup.mode = 'earliest') row format json;
3. get internal table id from dashboard, eg. __internal_s2_2_sourceinternaltable_1003
4. select * from __internal_s2_2_sourceinternaltable_1003

dev=> select * from __internal_s2_2_sourceinternaltable_1003 ;
 partition_id |                                               offset
--------------+-----------------------------------------------------------------------------------------------------
 0            |                                                                                                    +
              | \x05kafka\x12U{"topic":"kafka_2_partition_topic","partition":0,"start_offset":3,"stop_offset":null}
 0            |                                                                                                    +
              | \x05kafka\x12U{"topic":"kafka_2_partition_topic","partition":0,"start_offset":3,"stop_offset":null}
 0            |                                                                                                    +
              | \x05kafka\x12U{"topic":"kafka_2_partition_topic","partition":0,"start_offset":3,"stop_offset":null}
 0            |                                                                                                    +
              | \x05kafka\x12U{"topic":"kafka_2_partition_topic","partition":0,"start_offset":3,"stop_offset":null}
(4 rows)

./risedev d full
./scripts/source/prepare_ci_kafka.sh
psql -h localhost -p 4566 -d dev -U root -c "create table s2 (v1 int, v2 varchar) with ( connector = 'kafka', topic = 'kafka_2_partition_topic', properties.bootstrap.server = '127.0.0.1:29092', scan.startup.mode = 'earliest') row format json;"

Expected behavior

1 row

Additional context

None

[st1page]

"parallelism": 4 in batch plan.

explain (distsql) select * from "__internal_s2_2_sourceinternaltable_1003";
 {
   "root_stage_id": 0,
   "stages": {
     "1": {
       "root": {
         "plan_node_id": 20,
         "plan_node_type": "BatchSeqScan",
         "schema": [
           {
             "dataType": {
               "typeName": "VARCHAR",
               "isNullable": true
             },
             "name": "__internal_s2_2_sourceinternaltable_1003.partition_id"
           },
           {
             "dataType": {
               "typeName": "VARCHAR",
               "isNullable": true
             },
             "name": "__internal_s2_2_sourceinternaltable_1003.offset"
           }
         ],
         "children": [],
         "source_stage_id": null
       },
       "parallelism": 4,
       "exchange_info": {
         "mode": "SINGLE"
       }
     },
     "0": {
       "root": {
         "plan_node_id": 21,
         "plan_node_type": "BatchExchange",
         "schema": [
           {
             "dataType": {
               "typeName": "VARCHAR",
               "isNullable": true
             },
             "name": "__internal_s2_2_sourceinternaltable_1003.partition_id"
           },
           {
             "dataType": {
               "typeName": "VARCHAR",
               "isNullable": true
             },
             "name": "__internal_s2_2_sourceinternaltable_1003.offset"
           }
         ],
         "children": [],
         "source_stage_id": 1
       },
       "parallelism": 1,
       "exchange_info": {
         "mode": "SINGLE"
       }
     }
   },
   "child_edges": {
     "1": [],
     "0": [
       1
     ]
   },
   "parent_edges": {
     "1": [
       0
     ],

[BugenZhao]

After #5907, we will panic at The stage has single distribution, but contains a table scan node with multiple partitions. 😄

Actually, it's a little bit tricky to get the correct result. If we schedule multiple tasks, we'll get duplicated rows as there's no way to prune the distribution. If we schedule a single task, then it may behave strangely with #5850.

[tabVersion]

reopen this issue because the case is still there

Just checked on the latest main (25f6655)

[BugenZhao]

There seems something wrong with the distribution of BatchSeqScan::to_local: we should make it singleton for Source's internal tables. cc @kwannoel Would you please help to take a look?

[kwannoel]

There seems something wrong with the distribution of BatchSeqScan::to_local: we should make it singleton for Source's internal tables. cc @kwannoel Would you please help to take a look?

To provide some context, the distribution for BatchSeqScan on non-system tables is SomeShard after this PR: #7240

risingwave/src/frontend/src/optimizer/plan_node/batch_seq_scan.rs

Lines 258 to 261 in 1d2bb32

	} else {
	// NOTE(kwannoel): This is a hack to force an exchange to always be inserted before scan.
	Distribution::SomeShard
	};

By making the distribution SomeShard, an Exchange would be inserted.
This forces BatchSeqScan to be executed on the compute node instead of the frontend.

---

If BatchSeqScan uses SomeShard, it will just defer to table_scan_info to get number of partitions, and infer parallelism from there:

risingwave/src/frontend/src/scheduler/plan_fragmenter.rs

Lines 527 to 532 in 9f64e93

	if let Some(table_scan_info) = &table_scan_info {
	table_scan_info
	.partitions
	.as_ref()
	.map(|m| m.len())
	.unwrap_or(1)

Is it correct to assume that each worker thread should scan independent partitions?

And if so, there should be no duplicated results, even if Distribution::SomeShard? 🤔

[liurenjie1024]

Yes, I think this is caused by incorrect vnode mapping for internal table, and I'll take a look at this.

[tabVersion]

any updates?

[liurenjie1024]

I'll look into this recently.

[fuyufjh]

Any updates?

[liurenjie1024]

Will look into it later.

[tabVersion]

close as no updates