#[derive(Debug, Clone, Serialize, Deserialize)] struct WasmProcessorConfig {     /// WASM module path or inline WASM bytes (base64 encoded)     module: String,     /// Function name to call in the WASM module     function: String,     /// Whether the module field contains a file path or base64 encoded bytes     #[serde(default = "default_is_file")]     is_file: bool,+    /// Maximum memory pages (64KB each) the WASM module can use+    #[serde(default = "default_max_memory_pages")]+    max_memory_pages: u32,+    /// Timeout for WASM execution in milliseconds+    #[serde(default = "default_timeout_ms")]+    timeout_ms: u64, }

fndefault_max_memory_pages() ->u32{256// 16MB default limit}fndefault_timeout_ms() ->u64{5000// 5 seconds default timeout}

/// WasmProcessor executes WebAssembly modules for message processing.////// # WASM Module Requirements////// The WASM module must export:/// - `memory`: WebAssembly memory for data exchange/// - `alloc(size: i32) -> i32`: Allocate memory and return pointer/// - `dealloc(ptr: i32, size: i32)`: Free allocated memory (optional but recommended)/// - Processing function with signature `(ptr: i32, len: i32) -> i32`////// # Data Format////// Input: MessageBatch serialized as JSON bytes/// Output: Length-prefixed byte array (4-byte LE length + data)////// # Example WASM Module (Rust)/// ```ignore/// #[no_mangle]/// pub extern "C" fn alloc(size: i32) -> i32 {///     // Allocate memory/// }////// #[no_mangle]/// pub extern "C" fn process(ptr: i32, len: i32) -> i32 {///     // Process data and return output pointer/// }/// ```structWasmProcessor{

     fn message_batch_to_bytes(&self, msg_batch: &MessageBatch) -> Result<Vec<u8>, Error> {-        // Convert MessageBatch to JSON for WASM processing-        // This is a simplified approach - in practice you might want to use a more efficient format-        let json_data = msg_batch.to_binary("value")?;-        Ok(json_data.join(&b'\n'))+        // Serialize MessageBatch to JSON, preserving schema and metadata+        let json_value = serde_json::to_value(msg_batch)+            .map_err(|e| Error::Process(format!("Failed to serialize MessageBatch: {}", e)))?;+        serde_json::to_vec(&json_value)+            .map_err(|e| Error::Process(format!("Failed to convert to bytes: {}", e)))     }

         // Write input data to WASM memory         memory             .write(&mut store, input_ptr as usize, &input_data)             .map_err(|e| Error::Process(format!("Failed to write input data: {}", e)))?;          // Call the processing function         let output_ptr = func             .call(&mut store, (input_ptr, input_data.len() as i32))             .map_err(|e| Error::Process(format!("WASM function call failed: {}", e)))?;          // Read output data from WASM memory         let output_data = self.read_output_from_memory(&mut store, &memory, output_ptr)?;++        // Free allocated input memory+        if let Ok(dealloc_func) = instance.get_typed_func::<(i32, i32), ()>(&mut store, "dealloc") {+            dealloc_func+                .call(&mut store, (input_ptr, input_data.len() as i32))+                .map_err(|e| Error::Process(format!("Failed to deallocate memory: {}", e)))?;+        }

     fn bytes_to_message_batch(         &self,         data: Vec<u8>,         original: &MessageBatch,     ) -> Result<MessageBatch, Error> {-        // Convert processed bytes back to MessageBatch-        // This assumes the WASM function returns JSON data-        if data.is_empty() {-            return Ok(MessageBatch::new_binary(vec![]).unwrap());-        }--        // Try to parse as JSON and create new MessageBatch-        let json_str = String::from_utf8(data)-            .map_err(|e| Error::Process(format!("Invalid UTF-8 in WASM output: {}", e)))?;--        // For simplicity, return as binary data-        // In practice, you might want to parse JSON and convert back to Arrow format-        Ok(original.new_binary_with_origin(vec![json_str.into_bytes()])?)+        if data.is_empty() {+            return Ok(MessageBatch::new_binary(vec![]).unwrap());+        }++        // Deserialize JSON data back to MessageBatch+        let output_batch: MessageBatch = serde_json::from_slice(&data)+            .map_err(|e| Error::Process(format!("Failed to deserialize WASM output: {}", e)))?;++        Ok(output_batch)     }

         // Load WASM module-        let module_data =   // Load from file-            std::fs::read(&config.module).map_err(|e| {-                Error::Config(format!(-                    "Failed to read WASM file '{}': {}",-                    config.module, e-                ))-            })?;+        let module_data = if config.is_file {+            // Load from file+            std::fs::read(&config.module).map_err(|e| {+                Error::Config(format!(+                    "Failed to read WASM file '{}': {}",+                    config.module, e+                ))+            })?+        } else {+            // Decode from base64+            base64::engine::general_purpose::STANDARD+                .decode(&config.module)+                .map_err(|e| {+                    Error::Config(format!(+                        "Failed to decode base64 WASM module: {}",+                        e+                    ))+                })?+        };

 #[derive(Debug, Clone, Serialize, Deserialize)] struct WasmProcessorConfig {     /// WASM module path or inline WASM bytes (base64 encoded)     module: String,     /// Function name to call in the WASM module     function: String,+    /// Whether the module field contains a file path or base64 encoded bytes+    #[serde(default = "default_is_file")]+    is_file: bool, }++fn default_is_file() -> bool {+    true+}