in spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/RowBatch.java [196:532]
public void convertArrowToRowBatch() throws DorisException {
try {
for (int col = 0; col < fieldVectors.size(); col++) {
FieldVector curFieldVector = fieldVectors.get(col);
MinorType mt = curFieldVector.getMinorType();
final String colName = schema.get(col).getName();
final String currentType = schema.get(col).getType();
switch (currentType) {
case "NULL_TYPE":
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
addValueToRow(rowIndex, null);
}
break;
case "BOOLEAN":
Preconditions.checkArgument(mt.equals(MinorType.BIT),
typeMismatchMessage(colName, currentType, mt));
BitVector bitVector = (BitVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = bitVector.isNull(rowIndex) ? null : bitVector.get(rowIndex) != 0;
addValueToRow(rowIndex, fieldValue);
}
break;
case "TINYINT":
Preconditions.checkArgument(mt.equals(MinorType.TINYINT),
typeMismatchMessage(colName, currentType, mt));
TinyIntVector tinyIntVector = (TinyIntVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = tinyIntVector.isNull(rowIndex) ? null : tinyIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "SMALLINT":
Preconditions.checkArgument(mt.equals(MinorType.SMALLINT),
typeMismatchMessage(colName, currentType, mt));
SmallIntVector smallIntVector = (SmallIntVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = smallIntVector.isNull(rowIndex) ? null : smallIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "INT":
Preconditions.checkArgument(mt.equals(MinorType.INT),
typeMismatchMessage(colName, currentType, mt));
IntVector intVector = (IntVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = intVector.isNull(rowIndex) ? null : intVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "BIGINT":
Preconditions.checkArgument(mt.equals(MinorType.BIGINT),
typeMismatchMessage(colName, currentType, mt));
BigIntVector bigIntVector = (BigIntVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = bigIntVector.isNull(rowIndex) ? null : bigIntVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "LARGEINT":
Preconditions.checkArgument(mt.equals(MinorType.FIXEDSIZEBINARY) ||
mt.equals(MinorType.VARCHAR), typeMismatchMessage(colName, currentType, mt));
if (mt.equals(MinorType.FIXEDSIZEBINARY)) {
FixedSizeBinaryVector largeIntVector = (FixedSizeBinaryVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (largeIntVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
byte[] bytes = largeIntVector.get(rowIndex);
ArrayUtils.reverse(bytes);
BigInteger largeInt = new BigInteger(bytes);
addValueToRow(rowIndex, largeInt.toString());
}
} else {
VarCharVector largeIntVector = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (largeIntVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String stringValue = new String(largeIntVector.get(rowIndex));
addValueToRow(rowIndex, stringValue);
}
}
break;
case "IPV4":
Preconditions.checkArgument(mt.equals(MinorType.UINT4) || mt.equals(MinorType.INT) || mt.equals(MinorType.VARCHAR),
typeMismatchMessage(colName, currentType, mt));
if (mt.equals(MinorType.VARCHAR)) {
VarCharVector vector = (VarCharVector) curFieldVector;
for (int i = 0; i < rowCountInOneBatch; i++) {
addValueToRow(i, vector.isNull(i) ? null : new String(vector.get(i)));
}
} else {
BaseIntVector vector = (mt.equals(MinorType.INT)) ? (IntVector) curFieldVector : (UInt4Vector) curFieldVector;
for (int i = 0; i < rowCountInOneBatch; i++) {
addValueToRow(i, vector.isNull(i) ? null : IPUtils.convertLongToIPv4Address(vector.getValueAsLong(i)));
}
}
break;
case "FLOAT":
Preconditions.checkArgument(mt.equals(MinorType.FLOAT4),
typeMismatchMessage(colName, currentType, mt));
Float4Vector float4Vector = (Float4Vector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = float4Vector.isNull(rowIndex) ? null : float4Vector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "TIME":
case "DOUBLE":
Preconditions.checkArgument(mt.equals(MinorType.FLOAT8),
typeMismatchMessage(colName, currentType, mt));
Float8Vector float8Vector = (Float8Vector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = float8Vector.isNull(rowIndex) ? null : float8Vector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "BINARY":
Preconditions.checkArgument(mt.equals(MinorType.VARBINARY),
typeMismatchMessage(colName, currentType, mt));
VarBinaryVector varBinaryVector = (VarBinaryVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
Object fieldValue = varBinaryVector.isNull(rowIndex) ? null : varBinaryVector.get(rowIndex);
addValueToRow(rowIndex, fieldValue);
}
break;
case "DECIMAL":
Preconditions.checkArgument(mt.equals(MinorType.VARCHAR),
typeMismatchMessage(colName, currentType, mt));
VarCharVector varCharVectorForDecimal = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (varCharVectorForDecimal.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String decimalValue = new String(varCharVectorForDecimal.get(rowIndex));
Decimal decimal = new Decimal();
try {
decimal.set(new scala.math.BigDecimal(new BigDecimal(decimalValue)));
} catch (NumberFormatException e) {
String errMsg = "Decimal response result '" + decimalValue + "' is illegal.";
logger.error(errMsg, e);
throw new DorisException(errMsg);
}
addValueToRow(rowIndex, decimal);
}
break;
case "DECIMALV2":
case "DECIMAL32":
case "DECIMAL64":
case "DECIMAL128":
case "DECIMAL128I":
Preconditions.checkArgument(mt.equals(MinorType.DECIMAL),
typeMismatchMessage(colName, currentType, mt));
DecimalVector decimalVector = (DecimalVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (decimalVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
Decimal decimalV2 = Decimal.apply(decimalVector.getObject(rowIndex));
addValueToRow(rowIndex, decimalV2);
}
break;
case "DATE":
case "DATEV2":
Preconditions.checkArgument(mt.equals(MinorType.VARCHAR)
|| mt.equals(MinorType.DATEDAY), typeMismatchMessage(colName, currentType, mt));
if (mt.equals(MinorType.VARCHAR)) {
VarCharVector date = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (date.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String stringValue = new String(date.get(rowIndex));
LocalDate localDate = LocalDate.parse(stringValue);
if (datetimeJava8ApiEnabled) {
addValueToRow(rowIndex, localDate);
} else {
addValueToRow(rowIndex, Date.valueOf(localDate));
}
}
} else {
DateDayVector date = (DateDayVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (date.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
LocalDate localDate = LocalDate.ofEpochDay(date.get(rowIndex));
if (datetimeJava8ApiEnabled) {
addValueToRow(rowIndex, localDate);
} else {
addValueToRow(rowIndex, Date.valueOf(localDate));
}
}
}
break;
case "DATETIME":
case "DATETIMEV2":
if (mt.equals(MinorType.VARCHAR)) {
VarCharVector varCharVector = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (varCharVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String stringValue = completeMilliseconds(new String(varCharVector.get(rowIndex),
StandardCharsets.UTF_8));
LocalDateTime dateTime = LocalDateTime.parse(stringValue, dateTimeV2Formatter);
if (datetimeJava8ApiEnabled) {
Instant instant = dateTime.atZone(DEFAULT_ZONE_ID).toInstant();
addValueToRow(rowIndex, instant);
} else {
addValueToRow(rowIndex, Timestamp.valueOf(dateTime));
}
}
} else if (curFieldVector instanceof TimeStampVector) {
TimeStampVector timeStampVector = (TimeStampVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (timeStampVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
LocalDateTime dateTime = getDateTime(rowIndex, timeStampVector);
if (datetimeJava8ApiEnabled) {
Instant instant = dateTime.atZone(DEFAULT_ZONE_ID).toInstant();
addValueToRow(rowIndex, instant);
} else {
addValueToRow(rowIndex, Timestamp.valueOf(dateTime));
}
}
} else {
String errMsg = String.format("Unsupported type for DATETIMEV2, minorType %s, class is %s",
mt.name(), curFieldVector.getClass());
throw new java.lang.IllegalArgumentException(errMsg);
}
break;
case "CHAR":
case "VARCHAR":
case "STRING":
case "JSONB":
case "VARIANT":
Preconditions.checkArgument(mt.equals(MinorType.VARCHAR),
typeMismatchMessage(colName, currentType, mt));
VarCharVector varCharVector = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (varCharVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String value = new String(varCharVector.get(rowIndex), StandardCharsets.UTF_8);
addValueToRow(rowIndex, value);
}
break;
case "IPV6":
Preconditions.checkArgument(mt.equals(MinorType.VARCHAR),
typeMismatchMessage(colName, currentType, mt));
VarCharVector ipv6VarcharVector = (VarCharVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (ipv6VarcharVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
break;
}
// Compatible with IPv6 in Doris 2.1.3 and above.
String ipv6Str = new String(ipv6VarcharVector.get(rowIndex));
if (ipv6Str.contains(":")){
addValueToRow(rowIndex, ipv6Str);
}else {
String ipv6Address = IPUtils.fromBigInteger(new BigInteger(ipv6Str));
addValueToRow(rowIndex, ipv6Address);
}
}
break;
case "ARRAY":
Preconditions.checkArgument(mt.equals(MinorType.LIST),
typeMismatchMessage(colName, currentType, mt));
ListVector listVector = (ListVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (listVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String value = listVector.getObject(rowIndex).toString();
addValueToRow(rowIndex, value);
}
break;
case "MAP":
Preconditions.checkArgument(mt.equals(MinorType.MAP),
typeMismatchMessage(colName, currentType, mt));
MapVector mapVector = (MapVector) curFieldVector;
UnionMapReader reader = mapVector.getReader();
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (mapVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
reader.setPosition(rowIndex);
Map<String, String> value = new HashMap<>();
while (reader.next()) {
value.put(Objects.toString(reader.key().readObject(), null),
Objects.toString(reader.value().readObject(), null));
}
addValueToRow(rowIndex, value);
}
break;
case "STRUCT":
Preconditions.checkArgument(mt.equals(MinorType.STRUCT),
typeMismatchMessage(colName, currentType, mt));
StructVector structVector = (StructVector) curFieldVector;
for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
if (structVector.isNull(rowIndex)) {
addValueToRow(rowIndex, null);
continue;
}
String value = structVector.getObject(rowIndex).toString();
addValueToRow(rowIndex, value);
}
break;
default:
String errMsg = "Unsupported type " + schema.get(col).getType() + "of col: " + schema.get(col).getName();
logger.error(errMsg);
throw new DorisException(errMsg);
}
}
} catch (Exception e) {
close();
throw e;
}
}