feat(dal-sink): support dynamic json handle mode

src/connector/src/sink/file_sink/opendal_sink.rs

@@ -152,7 +152,7 @@ impl<S: OpendalSinkBackend> Sink for FileSink<S> {
             &self.path,
             self.schema.clone(),
             writer_param.executor_id,
-            self.format_desc.encode.clone(),
+            &self.format_desc,
             self.engine_type.clone(),
             self.batching_strategy.clone(),
         )?;
@@ -350,28 +350,28 @@ impl OpenDalSinkWriter {
         write_path: &str,
         rw_schema: Schema,
         executor_id: u64,
-        encode_type: SinkEncode,
+        format_desc: &SinkFormatDesc,
         engine_type: EngineType,
         batching_strategy: BatchingStrategy,
     ) -> Result<Self> {
         let arrow_schema = convert_rw_schema_to_arrow_schema(rw_schema.clone())?;
+        let jsonb_handling_mode = JsonbHandlingMode::from_options(&format_desc.options)?;
         let row_encoder = JsonEncoder::new(
             rw_schema,
             None,
             crate::sink::encoder::DateHandlingMode::String,
             TimestampHandlingMode::String,
             TimestamptzHandlingMode::UtcString,
             TimeHandlingMode::String,
-            JsonbHandlingMode::String,
+            jsonb_handling_mode,
         );
         Ok(Self {
             schema: Arc::new(arrow_schema),
             write_path: write_path.to_owned(),
             operator,
             sink_writer: None,
             executor_id,
-
-            encode_type,
+            encode_type: format_desc.encode.clone(),
             row_encoder,
             engine_type,
             batching_strategy,