TyrantLucifer commented on code in PR #2750:
URL:
https://github.com/apache/incubator-seatunnel/pull/2750#discussion_r973594479
##########
seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/common/serialization/InternalRowConverter.java:
##########
@@ -213,4 +225,60 @@ private static SeaTunnelRow reconvert(InternalRow
engineRow, SeaTunnelRowType ro
}
return new SeaTunnelRow(fields);
}
+
+ private static DataType seaTunnelType2SparkType(SeaTunnelDataType<?>
seaTunnelDataType) {
+ SqlType sqlType = seaTunnelDataType.getSqlType();
+ switch (sqlType) {
+ case ARRAY:
+ BasicType<?> elementType = ((ArrayType<?, ?>)
seaTunnelDataType).getElementType();
+ return
DataTypes.createArrayType(seaTunnelType2SparkType(elementType));
+ case MAP:
+ SeaTunnelDataType<?> keyType = ((MapType<?, ?>)
seaTunnelDataType).getKeyType();
+ SeaTunnelDataType<?> valueType = ((MapType<?, ?>)
seaTunnelDataType).getValueType();
+ return
DataTypes.createMapType(seaTunnelType2SparkType(keyType),
seaTunnelType2SparkType(valueType));
+ case STRING:
+ return DataTypes.StringType;
+ case BOOLEAN:
+ return DataTypes.BooleanType;
+ case TINYINT:
+ return DataTypes.ByteType;
+ case SMALLINT:
+ return DataTypes.ShortType;
+ case INT:
+ return DataTypes.IntegerType;
+ case BIGINT:
+ return DataTypes.LongType;
+ case FLOAT:
+ return DataTypes.FloatType;
+ case DOUBLE:
+ return DataTypes.DoubleType;
+ case DECIMAL:
+ int precision = ((DecimalType)
seaTunnelDataType).getPrecision();
+ int scale = ((DecimalType) seaTunnelDataType).getScale();
+ return DataTypes.createDecimalType(precision, scale);
+ case NULL:
+ return DataTypes.NullType;
+ case BYTES:
+ return DataTypes.BinaryType;
+ case DATE:
+ return DataTypes.DateType;
+ case TIMESTAMP:
+ return DataTypes.TimestampType;
+ case TIME:
+ throw new RuntimeException("SeaTunnel not support time type,
it will be supported in the future");
+ case ROW:
+ ArrayList<StructField> structFields = new ArrayList<>();
+ SeaTunnelDataType<?>[] fieldTypes = ((SeaTunnelRowType)
seaTunnelDataType).getFieldTypes();
+ String[] fieldNames = ((SeaTunnelRowType)
seaTunnelDataType).getFieldNames();
+ for (int i = 0; i < fieldNames.length; i++) {
+ StructField structField = new StructField(fieldNames[i],
seaTunnelType2SparkType(fieldTypes[i]), true, null);
+ structFields.add(structField);
+ }
+ return DataTypes.createStructType(structFields);
+ default:
+ // do nothing
+ // never get in there
+ return null;
+ }
+ }
Review Comment:
Sorry, I don't see the code for this piece. I will correct it immediately
after
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
