This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch fixDataTypeConversionForSparkIoTDBTableConnector in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit 68686a294568ec117ee1834926b5ab8b64bc1912 Author: shuwenwei <[email protected]> AuthorDate: Thu Apr 24 16:20:36 2025 +0800 fix data type conversion for spark iotdb table connector --- .../scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala | 8 +++++++- .../apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala | 11 ++++++++--- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala b/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala index 97ea02e..c640caa 100644 --- a/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala +++ b/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) + } } } diff --git a/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala b/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala index 91f921d..c2feb29 100644 --- a/connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala +++ b/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)
