diff --git a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeRandomData.java b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeRandomData.java
index 9a8aa1888a9..ff18abd9557 100644
--- a/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeRandomData.java
+++ b/seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeRandomData.java
@@ -72,7 +72,7 @@ private Object randomColumnValue(SeaTunnelDataType<?> fieldType) {
         if (BOOLEAN_TYPE.equals(fieldType)) {
             return RandomUtils.nextInt(0, 2) == 1;
         } else if (BYTE_TYPE.equals(fieldType)) {
-            return (byte) RandomUtils.nextInt(0, Byte.MAX_VALUE);
+            return (byte) RandomUtils.nextInt(Byte.MIN_VALUE, Byte.MAX_VALUE);
         } else if (SHORT_TYPE.equals(fieldType)) {
             return (short) RandomUtils.nextInt(Byte.MAX_VALUE, Short.MAX_VALUE);
         } else if (INT_TYPE.equals(fieldType)) {
@@ -94,7 +94,7 @@ private Object randomColumnValue(SeaTunnelDataType<?> fieldType) {
         } else if (fieldType instanceof DecimalType) {
             DecimalType decimalType = (DecimalType) fieldType;
             return new BigDecimal(RandomStringUtils.randomNumeric(decimalType.getPrecision() - decimalType.getScale()) + "." +
-                RandomStringUtils.randomNumeric(decimalType.getPrecision() - decimalType.getScale()));
+                RandomStringUtils.randomNumeric(decimalType.getScale()));
         } else if (fieldType instanceof ArrayType) {
             ArrayType<?, ?> arrayType = (ArrayType<?, ?>) fieldType;
             BasicType<?> elementType = arrayType.getElementType();
@@ -116,7 +116,7 @@ private Object randomColumnValue(SeaTunnelDataType<?> fieldType) {
         } else if (VOID_TYPE.equals(fieldType) || fieldType == null) {
             return Void.TYPE;
         } else {
-            throw new IllegalStateException("Unexpected value: " + fieldType);
+            throw new UnsupportedOperationException("Unexpected value: " + fieldType);
         }
     }
 
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/assertion/fakesource_to_assert.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/assertion/fakesource_to_assert.conf
index d1e9c45839c..350970f9b60 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/assertion/fakesource_to_assert.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/assertion/fakesource_to_assert.conf
@@ -25,64 +25,66 @@ env {
   #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
 }
 
+# Docoment-Link
+# https://seatunnel.apache.org/docs/category/source-v2
+# https://seatunnel.apache.org/docs/category/sink-v2
+# https://seatunnel.apache.org/docs/category/transform
+
 source {
-  # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name,age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+    result_table_name = "fake"
+  }
 }
 
-transform {
-    sql {
-      sql = "select name,age from fake"
-    }
 
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+transform {
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 }
 
 sink {
-   Assert {
-       rules =
-           [{
-               field_name = name
-               field_type = string
-               field_value = [
-                   {
-                       rule_type = NOT_NULL
-                   },
-                   {
-                       rule_type = MIN_LENGTH
-                       rule_value = 3
-                   },
-                   {
-                        rule_type = MAX_LENGTH
-                        rule_value = 20
-                   }
-               ]
-           },{
-               field_name = age
-               field_type = int
-               field_value = [
-                   {
-                       rule_type = NOT_NULL
-                   },
-                   {
-                       rule_type = MIN
-                       rule_value = 1
-                   },
-                   {
-                        rule_type = MAX
-                        rule_value = 100
-                   }
-               ]
-           }
-           ]
-   }
+  Assert {
+    rules = [
+      {
+        field_name = c_boolean
+        field_type = boolean
+        field_value = [
+          {
+            rule_type = NOT_NULL
+          }
+        ]
+      },
+      {
+        field_name = c_string
+        field_type = string
+        field_value = [
+          {
+            rule_type = NOT_NULL
+          }
+        ]
+      }
+    ]
+  }
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Assert
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
index 49be0920fe8..18b09b10d3a 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
@@ -27,28 +27,37 @@ env {
 }
 
 source {
-  # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name,age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+    result_table_name = "fake"
+  }
 }
 
 transform {
-    sql {
-      sql = "select name,age from fake"
-    }
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   Console {}
-
-  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
index 769c8760de5..5174299653c 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
@@ -28,38 +28,51 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="json"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "json"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
index 9f5fd0b1787..2c5aecf98d9 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
@@ -28,39 +28,52 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="parquet"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "parquet"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
index ef83dfd4e4a..accab04da4d 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
@@ -28,39 +28,52 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="text"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "text"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
index b18b472f6cd..c0073edc01d 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
@@ -28,38 +28,51 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="json"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "json"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/LocalFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
index 9e2d5ad96af..f37728240f6 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
@@ -28,39 +28,51 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="parquet"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "parquet"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/LocalFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
index d162b101f0c..bc28a5c540a 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
@@ -28,39 +28,52 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="text"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["age"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "text"
+    sink_columns = ["name", "age"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/LocalFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
index da1ae493678..5593e5f7c71 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
@@ -19,41 +19,54 @@
 ######
 
 env {
-    # You can set flink configuration here
-    execution.parallelism = 1
-    job.mode = "BATCH"
-    #execution.checkpoint.interval = 10000
-    #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+  # You can set flink configuration here
+  execution.parallelism = 1
+  job.mode = "BATCH"
+  #execution.checkpoint.interval = 10000
+  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
 }
 
 source {
-    FakeSource {
-        result_table_name = "fake"
-        field_name = "name, age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
-    # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-    # please go to https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
+    result_table_name = "fake"
+  }
 }
 
-transform {
-    sql {
-        sql = "select * from (values('root.ln.d1', '1660147200000', 'status,value', 'true,1001'), ('root.ln.d1', '1660233600000', 'status,value', 'false,1002')) t (device, `timestamp`, measurements, `values`)"
-    }
 
-    # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-    # please go to https://seatunnel.apache.org/docs/transform/sql
+transform {
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 }
 
 sink {
-    IoTDB {
-        node_urls = ["flink_e2e_iotdb_sink:6667"]
-        username = "root"
-        password = "root"
-        batch_size = 1
-        batch_interval_ms = 10
-    }
+  IoTDB {
+    node_urls = ["flink_e2e_iotdb_sink:6667"]
+    username = "root"
+    password = "root"
+    batch_size = 1
+    batch_interval_ms = 10
+  }
 
-    # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-    # please go to https://seatunnel.apache.org/docs/connector-v2/sink/IoTDB
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/IoTDB
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/fakesource_to_jdbc.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/fakesource_to_jdbc.conf
index 9640e19c228..5b4d959aec5 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/fakesource_to_jdbc.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/fakesource_to_jdbc.conf
@@ -27,35 +27,46 @@ env {
 }
 
 source {
-  # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+    result_table_name = "fake"
+  }
 }
 
-transform {
-    sql {
-      sql = "select name from fake"
-    }
 
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+transform {
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 }
-
 sink {
   Jdbc {
-      source_table_name = fake
-      driver = org.postgresql.Driver
-      url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
-      user = test
-      password = test
-      query = "insert into test(name) values(?)"
+    source_table_name = fake
+    driver = org.postgresql.Driver
+    url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
+    user = test
+    password = test
+    query = "insert into test(name) values(?)"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/jdbcsource_to_console.conf b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/jdbcsource_to_console.conf
index 6862abc04c2..3586dcacce0 100644
--- a/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/jdbcsource_to_console.conf
+++ b/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/jdbc/jdbcsource_to_console.conf
@@ -27,7 +27,6 @@ env {
 }
 
 source {
-  # This is a example source plugin **only for test and demonstrate the feature source plugin**
     Jdbc {
         driver = org.postgresql.Driver
         url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF"
@@ -35,19 +34,11 @@ source {
         password = "test"
         query = "select * from test"
     }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
 }
 
 transform {
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   Console {}
-  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
index 7b812762051..5762efc941a 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/fake/fakesource_to_console.conf
@@ -30,44 +30,37 @@ env {
 }
 
 source {
-  # This is a example input plugin **only for test and demonstrate the feature input plugin**
   FakeSource {
-    result_table_name = "my_dataset"
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
+    result_table_name = "fake"
   }
-
-  # You can also use other input plugins, such as hdfs
-  # hdfs {
-  #   result_table_name = "accesslog"
-  #   path = "hdfs://hadoop-cluster-01/nginx/accesslog"
-  #   format = "json"
-  # }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of input plugins,
-  # please go to https://seatunnel.apache.org/docs/spark/configuration/source-plugins/Fake
 }
 
-transform {
-  # split data by specific delimiter
-
-  # you can also use other transform plugins, such as sql
-  # sql {
-  #   sql = "select * from accesslog where request_time > 1000"
-  # }
 
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/spark/configuration/transform-plugins/Split
+transform {
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 }
 
 sink {
-  # choose stdout output plugin to output data to console
   Console {}
-
-  # you can also you other output plugins, such as sql
-  # hdfs {
-  #   path = "hdfs://hadoop-cluster-01/nginx/accesslog_processed"
-  #   save_mode = "append"
-  # }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of output plugins,
-  # please go to https://seatunnel.apache.org/docs/spark/configuration/sink-plugins/Console
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
index c4d1aabe59b..32dc2917eca 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_json.conf
@@ -27,38 +27,51 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="json"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "json"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
index bdae80d74d1..0a0d4302e3a 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_parquet.conf
@@ -27,39 +27,52 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="parquet"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "parquet"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
index b682d3831cf..b04f98d4670 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_text.conf
@@ -27,39 +27,51 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   HdfsFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="text"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "text"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/HdfsFile
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
index d257f81bb44..4a10db0a550 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_json.conf
@@ -27,38 +27,48 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="json"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "json"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/File
 }
\ No newline at end of file
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
index b5d1412120a..464c249f26c 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_parquet.conf
@@ -27,37 +27,50 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Fake
 }
 
+
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/transform/sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="parquet"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "parquet"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
index 733a48e61a5..fdf54a60408 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_text.conf
@@ -27,37 +27,49 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
 }
 
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
   }
-
-  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
 }
 
 sink {
   LocalFile {
-    path="/tmp/hive/warehouse/test2"
-    field_delimiter="\t"
-    row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
-    file_name_expression="${transactionId}_${now}"
-    file_format="text"
-    sink_columns=["name","age"]
-    filename_time_format="yyyy.MM.dd"
-    is_enable_transaction=true
-    save_mode="error"
+    path = "/tmp/hive/warehouse/test2"
+    field_delimiter = "\t"
+    row_delimiter = "\n"
+    partition_by = ["c_bigint"]
+    partition_dir_expression = "${k0}=${v0}"
+    is_partition_field_write_in_file = true
+    file_name_expression = "${transactionId}_${now}"
+    file_format = "text"
+    sink_columns = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
+    filename_time_format = "yyyy.MM.dd"
+    is_enable_transaction = true
+    save_mode = "error"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
diff --git a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
index 9c7e521b73d..c8e2228180a 100644
--- a/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
+++ b/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/iotdb/fakesource_to_iotdb.conf
@@ -16,43 +16,55 @@
 #
 
 env {
-    # You can set spark configuration here
-    spark.app.name = "SeaTunnel"
-    spark.executor.instances = 2
-    spark.executor.cores = 1
-    spark.executor.memory = "1g"
-    spark.master = local
-    job.mode = "BATCH"
+  # You can set spark configuration here
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 2
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
+  job.mode = "BATCH"
 }
 
 source {
-    FakeSource {
-        result_table_name = "fake"
-        field_name = "name, age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
-    # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-    # please go to https://seatunnel.apache.org/docs/connector-v2/source/FakeSource
+    result_table_name = "fake"
+  }
 }
 
 transform {
-    sql {
-        sql = "select * from (values('root.ln.d1', '1660147200000', 'status,value', 'true,1001'), ('root.ln.d1', '1660233600000', 'status,value', 'false,1002')) t (device, `timestamp`, measurements, `values`)"
-    }
-
-    # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-    # please go to https://seatunnel.apache.org/docs/transform/sql
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+    result_table_name = "sql"
+  }
 }
 
 sink {
-    IoTDB {
-        node_urls = ["spark_e2e_iotdb_sink:6668"]
-        username = "root"
-        password = "root"
-        batch_size = 1
-        batch_interval_ms = 10
-    }
+  IoTDB {
+    node_urls = ["spark_e2e_iotdb_sink:6668"]
+    username = "root"
+    password = "root"
+    batch_size = 1
+    batch_interval_ms = 10
+  }
 
-    # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-    # please go to https://seatunnel.apache.org/docs/connector-v2/sink/IoTDB
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/IoTDB
 }
\ No newline at end of file
diff --git a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_console.conf b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_console.conf
index feb394127ac..c9a8af0d8a9 100644
--- a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_console.conf
+++ b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_console.conf
@@ -28,27 +28,44 @@ env {
 
 source {
   # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name,age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
+    result_table_name = "fake"
+  }
   # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+  # please go to https://seatunnel.apache.org/docs/category/source-v2
 }
 
 transform {
     sql {
-      sql = "select name,age from fake"
+      sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+      result_table_name = "sql"
     }
 
   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+  # please go to https://seatunnel.apache.org/docs/category/transform
 }
 
 sink {
   Console {}
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+  # please go to https://seatunnel.apache.org/docs/category/sink-v2
 }
\ No newline at end of file
diff --git a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_dingtalk.conf b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_dingtalk.conf
index aed13429043..e7c2cc91aa6 100644
--- a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_dingtalk.conf
+++ b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fake_to_dingtalk.conf
@@ -28,22 +28,38 @@ env {
 
 source {
   # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name,age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
+    result_table_name = "fake"
+  }
   # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+  # please go to https://seatunnel.apache.org/docs/category/source-v2
 }
 
 transform {
-    sql {
-      sql = "select name,age from fake"
-    }
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+  }
 
   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+  # please go to https://seatunnel.apache.org/docs/category/transform
 }
 
 sink {
@@ -53,5 +69,5 @@ sink {
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+  # please go to https://seatunnel.apache.org/docs/category/sink-v2
 }
\ No newline at end of file
diff --git a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fakesource_to_file.conf b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fakesource_to_file.conf
index f7b790c40b8..9a6589af1ca 100644
--- a/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fakesource_to_file.conf
+++ b/seatunnel-examples/seatunnel-flink-connector-v2-example/src/main/resources/examples/fakesource_to_file.conf
@@ -28,22 +28,38 @@ env {
 
 source {
   # This is a example source plugin **only for test and demonstrate the feature source plugin**
-    FakeSource {
-      result_table_name = "fake"
-      field_name = "name,age"
+  FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
     }
-
+    result_table_name = "fake"
+  }
   # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+  # please go to https://seatunnel.apache.org/docs/category/source-v2
 }
 
 transform {
+  sql {
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
+  }
 
-    sql {
-      sql = "select name,age from fake"
-    }
   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+  # please go to https://seatunnel.apache.org/docs/category/transform
 }
 
 sink {
@@ -64,5 +80,5 @@ sink {
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
-  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
+  # please go to https://seatunnel.apache.org/docs/category/sink-v2
 }
\ No newline at end of file
diff --git a/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.clickhouse.conf b/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.clickhouse.conf
index 2e27410ee86..5c0654511b8 100644
--- a/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.clickhouse.conf
+++ b/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.clickhouse.conf
@@ -25,17 +25,42 @@ env {
 
 source {
   FakeSource {
+    fields {
+      c_map = "map<string, string>"
+      c_array = "array<tinyint>"
+      c_string = string
+      c_boolean = boolean
+      c_tinyint = tinyint
+      c_smallint = smallint
+      c_int = int
+      c_bigint = bigint
+      c_float = float
+      c_double = double
+      c_decimal = "decimal(30, 8)"
+      c_null = "null"
+      c_bytes = bytes
+      c_date = date
+      c_time = time
+      c_timestamp = timestamp
+    }
     result_table_name = "fake"
-    field_name = "name,age"
   }
+  # If you would like to get more information about how to configure seatunnel and see full list of input plugins,
+  # please go to https://seatunnel.apache.org/docs/category/source-v2
 }
 
 
 transform {
+  # split data by specific delimiter
+
+  # you can also use other transform plugins, such as sql
   sql {
-    sql = "select name,age from fake"
+    sql = "select c_string,c_boolean,c_tinyint,c_smallint,c_int,c_bigint,c_float,c_double from fake"
     result_table_name = "sql"
   }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/category/transform
 }
 
 sink {
@@ -43,7 +68,7 @@ sink {
     host = "139.198.158.103:8123"
     database = "default"
     table = "test_clickhouse_table_v2"
-    fields = ["name", "age"]
+    fields = ["c_string", "c_boolean", "c_tinyint", "c_smallint", "c_int", "c_bigint", "c_float", "c_double"]
     username = 'default'
     bulk_size = 20000
     retry_codes = [209, 210]
diff --git a/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.conf b/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.conf
index ddd664f3aa7..5784d9345a0 100644
--- a/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.conf
+++ b/seatunnel-examples/seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.conf
@@ -62,7 +62,7 @@ source {
   # }
 
   # If you would like to get more information about how to configure seatunnel and see full list of input plugins,
-  # please go to https://seatunnel.apache.org/docs/2.1.3/category/source
+  # please go to https://seatunnel.apache.org/docs/category/source-v2
 }
 
 transform {
@@ -75,7 +75,7 @@ transform {
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
-  # please go to https://seatunnel.apache.org/docs/2.1.3/category/transform
+  # please go to https://seatunnel.apache.org/docs/category/transform
 }
 
 sink {
@@ -89,5 +89,5 @@ sink {
   # }
 
   # If you would like to get more information about how to configure seatunnel and see full list of output plugins,
-  # please go to https://seatunnel.apache.org/docs/2.1.3/category/sink
+  # please go to https://seatunnel.apache.org/docs/category/sink-v2
 }