fix compilation with new Executor trait

src/stream/src/executor/source/kafka_backfill_executor.rs

@@ -118,7 +118,7 @@ impl BackfillState {
 pub struct KafkaBackfillExecutor<S: StateStore> {
     pub inner: KafkaBackfillExecutorInner<S>,
     /// Upstream changelog stream which may contain metadata columns, e.g. `_rw_offset`
-    pub input: Box<dyn Executor>,
+    pub input: Executor,
 }
 
 pub struct KafkaBackfillExecutorInner<S: StateStore> {
@@ -222,7 +222,7 @@ impl<S: StateStore> KafkaBackfillExecutorInner<S> {
     }
 
     #[try_stream(ok = Message, error = StreamExecutorError)]
-    async fn execute(mut self, input: BoxedExecutor) {
+    async fn execute(mut self, input: Executor) {
         let mut input = input.execute();
 
         // Poll the upstream to get the first barrier.
@@ -805,22 +805,10 @@ fn compare_kafka_offset(a: &str, b: &str) -> Ordering {
     a.cmp(&b)
 }
 
-impl<S: StateStore> Executor for KafkaBackfillExecutor<S> {
+impl<S: StateStore> Execute for KafkaBackfillExecutor<S> {
     fn execute(self: Box<Self>) -> BoxedMessageStream {
         self.inner.execute(self.input).boxed()
     }
-
-    fn schema(&self) -> &Schema {
-        &self.inner.info.schema
-    }
-
-    fn pk_indices(&self) -> PkIndicesRef<'_> {
-        &self.inner.info.pk_indices
-    }
-
-    fn identity(&self) -> &str {
-        &self.inner.info.identity
-    }
 }
 
 impl<S: StateStore> Debug for KafkaBackfillExecutorInner<S> {

src/stream/src/from_proto/source_backfill.rs

@@ -37,7 +37,7 @@ impl ExecutorBuilder for KafkaBackfillExecutorBuilder {
         params: ExecutorParams,
         node: &Self::Node,
         store: impl StateStore,
-    ) -> StreamResult<BoxedExecutor> {
+    ) -> StreamResult<Executor> {
         let [input]: [_; 1] = params.input.try_into().unwrap();
 
         // let (sender, barrier_receiver) = unbounded_channel();
@@ -103,6 +103,7 @@ impl ExecutorBuilder for KafkaBackfillExecutorBuilder {
 
         let source_ctrl_opts = SourceCtrlOpts {
             chunk_size: params.env.config().developer.chunk_size,
+            rate_limit: None,
         };
 
         let source_column_ids: Vec<_> = source_columns
@@ -140,6 +141,10 @@ impl ExecutorBuilder for KafkaBackfillExecutorBuilder {
             params.env.connector_params(),
             backfill_state_table,
         );
-        Ok(KafkaBackfillExecutor { inner: exec, input }.boxed())
+        Ok((
+            params.info,
+            KafkaBackfillExecutor { inner: exec, input }.boxed(),
+        )
+            .into())
     }
 }