[AURON #1848] Introduce native protobuf deserializer

Cargo.toml

@@ -176,6 +176,8 @@ parking_lot = "0.12.5"
 paste = "1.0.15"
 procfs = "0.18.0"
 prost = "0.14.3"
+prost-types = "0.14.3"
+prost-reflect = "0.16.3"
 rand = "0.9.2"
 smallvec = "2.0.0-alpha.11"
 sonic-rs = "0.5.7"

native-engine/datafusion-ext-plans/Cargo.toml

@@ -63,6 +63,9 @@ paste = { workspace = true }
 smallvec = { workspace = true }
 tokio = { workspace = true }
 unchecked-index = { workspace = true }
+prost = { workspace = true }
+prost-types = { workspace = true }
+prost-reflect = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs = { workspace = true }

native-engine/datafusion-ext-plans/src/flink/mod.rs

@@ -0,0 +1,16 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod serde;

native-engine/datafusion-ext-plans/src/flink/serde/flink_deserializer.rs

@@ -0,0 +1,30 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use arrow::array::{BinaryArray, Int32Array, Int64Array, RecordBatch};
+
+/// FlinkDeserializer is used to deserialize messages from kafka.
+/// Supports Protobuf, JSON, etc.
+pub trait FlinkDeserializer {
+    /// Parse messages from kafka, including kafka metadata such as partitions,
+    /// offsets, and timestamps.
+    fn parse_messages_with_kafka_meta(
+        &mut self,
+        messages: &BinaryArray,
+        kafka_partition: &Int32Array,
+        kafka_offset: &Int64Array,
+        kafka_timestamp: &Int64Array,
+    ) -> datafusion::common::Result<RecordBatch>;
+}

native-engine/datafusion-ext-plans/src/flink/serde/mod.rs

@@ -0,0 +1,21 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod flink_deserializer;
+pub mod pb_deserializer;
+pub mod shared_array_builder;
+pub mod shared_list_array_builder;
+pub mod shared_map_array_builder;
+pub mod shared_struct_array_builder;