fix data type conversion for spark iotdb table connector

Author: shuwenwei

connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala

@@ -127,7 +127,13 @@ object IoTDBUtils {
       case BinaryType => TSDataType.BLOB
       case DateType => TSDataType.DATE
       case TimestampType => TSDataType.STRING
-      case _ => TSDataType.STRING
+      case _ => {
+        var errMsg = s"Unable to convert Spark data type $sparkDataType to IoTDB data type."
+        if (sparkDataType.simpleString.toLowerCase.contains("decimal")) {
+          errMsg += s"For float numbers in insert into values sql, you should add the suffix 'f' or 'd' to represent float or double."
+        }
+        throw new IllegalArgumentException(errMsg)
+      }
     }
   }
 

connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala

@@ -25,7 +25,7 @@ import org.apache.spark.SparkException
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.connector.write.{DataWriter, WriterCommitMessage}
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{NullType, StructType}
 import org.apache.tsfile.enums.TSDataType
 import org.apache.tsfile.write.record.Tablet
 import org.apache.tsfile.write.record.Tablet.ColumnCategory
@@ -57,10 +57,15 @@ class IoTDBDataWriter(options: IoTDBOptions, writeSchema: StructType, tableSchem
       } else {
         tableSchema.fields(i)
       }
-      val columnCategoryStr = tableSchemaMap.getOrElse(fieldInTableSchema.name, tableSchema.fields(i)).metadata.getString(IoTDBUtils.COLUMN_CATEGORY)
+      val actualColumnSchema = tableSchemaMap.getOrElse(fieldInTableSchema.name, tableSchema.fields(i))
+      val columnCategoryStr = actualColumnSchema.metadata.getString(IoTDBUtils.COLUMN_CATEGORY)
       val columnCategory = IoTDBUtils.getIoTDBColumnCategory(columnCategoryStr)
       if (fieldInTableSchema.name != IoTDBUtils.TIME) {
-        val dataType = writeSchemaField.dataType
+        val dataType = if (writeSchemaField.dataType == NullType) {
+          actualColumnSchema.dataType
+        } else {
+          writeSchemaField.dataType
+        }
         columnNameList.add(fieldInTableSchema.name)
         dataTypeList.add(IoTDBUtils.getIoTDBDataType(dataType))
         columnCategoryList.add(columnCategory)