add assertion

Signed-off-by: Richard Chien <[email protected]>

src/stream/src/task/stream_manager.rs

@@ -568,10 +568,10 @@ impl LocalStreamManagerCore {
         // Build the executor with params.
         let executor_params = ExecutorParams {
             env: env.clone(),
-            pk_indices,
+            pk_indices: pk_indices.clone(),
             executor_id,
             operator_id,
-            identity,
+            identity: identity.clone(),
             op_info,
             schema,
             input,
@@ -583,6 +583,11 @@ impl LocalStreamManagerCore {
         };
 
         let executor = create_executor(executor_params, self, node, store).await?;
+        assert_eq!(
+            executor.pk_indices(),
+            &pk_indices,
+            "`pk_indices` of {identity} not consistent with what derived by optimizer"
+        );
 
         // Wrap the executor for debug purpose.
         let executor = WrapperExecutor::new(