fix(sink): pass downstream pk when starting a sink (risingwavelabs#8660)

Author: xx01cyx

ci/scripts/e2e-iceberg-sink-test.sh

@@ -86,7 +86,7 @@ spark-3.3.1-bin-hadoop3/bin/spark-sql --packages $DEPENDENCIES \
 
 # check sink destination using shell
 if cat ./spark-output/*.csv | sort | awk -F "," '{
-if ($1 == 1 && $2 == 2 && $3 == "1-2") c1++;
+if ($1 == 1 && $2 == 50 && $3 == "1-50") c1++;
  if ($1 == 13 && $2 == 2 && $3 == "13-2") c2++;
   if ($1 == 21 && $2 == 2 && $3 == "21-2") c3++;
    if ($1 == 2 && $2 == 2 && $3 == "2-2") c4++;

e2e_test/sink/iceberg_sink.slt

@@ -23,6 +23,12 @@ INSERT INTO t6 VALUES (1, 2, '1-2'), (2, 2, '2-2'), (3, 2, '3-2'), (5, 2, '5-2')
 statement ok
 FLUSH;
 
+statement ok
+INSERT INTO t6 VALUES (1, 50, '1-50');
+
+statement ok
+FLUSH;
+
 statement ok
 DROP SINK s6;
 

src/stream/src/from_proto/sink.rs

@@ -37,10 +37,10 @@ impl ExecutorBuilder for SinkExecutorBuilder {
         let sink_type = SinkType::from_proto(sink_desc.get_sink_type().unwrap());
         let mut properties = sink_desc.get_properties().clone();
         let pk_indices = sink_desc
-            .plan_pk
+            .downstream_pk
             .iter()
-            .map(|pk| pk.column_index as usize)
-            .collect::<Vec<_>>();
+            .map(|i| *i as usize)
+            .collect_vec();
         let schema = sink_desc.columns.iter().map(Into::into).collect();
         // This field can be used to distinguish a specific actor in parallelism to prevent
         // transaction execution errors