You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by di...@apache.org on 2023/01/31 08:44:04 UTC
[doris-flink-connector] branch master updated: [Feature](Type)add new type of doris-flink-connector (#98)
This is an automated email from the ASF dual-hosted git repository.
diwu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new a979fb3 [Feature](Type)add new type of doris-flink-connector (#98)
a979fb3 is described below
commit a979fb3932518717cfe6e90e4c33274cb7ff396a
Author: lsy3993 <11...@users.noreply.github.com>
AuthorDate: Tue Jan 31 16:44:00 2023 +0800
[Feature](Type)add new type of doris-flink-connector (#98)
---
.../apache/doris/flink/catalog/DorisTypeMapper.java | 12 ++++++++++++
.../apache/doris/flink/serialization/RowBatch.java | 19 +++++++++++++++++++
.../src/main/thrift/doris/Types.thrift | 13 +++++++++++--
3 files changed, 42 insertions(+), 2 deletions(-)
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java
index f7a16ce..4805704 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/catalog/DorisTypeMapper.java
@@ -28,6 +28,10 @@ public class DorisTypeMapper {
private static final String DORIS_BIGINT = "BIGINT";
private static final String DORIS_LARGEINT = "BIGINT UNSIGNED";
private static final String DORIS_DECIMAL = "DECIMAL";
+ private static final String DORIS_DECIMALV2 = "DECIMALV2";
+ private static final String DORIS_DECIMAL32 = "DECIMAL32";
+ private static final String DORIS_DECIMAL64 = "DECIMAL64";
+ private static final String DORIS_DECIMAL128I = "DECIMAL128I";
private static final String DORIS_FLOAT = "FLOAT";
private static final String DORIS_DOUBLE = "DOUBLE";
@@ -39,7 +43,9 @@ public class DorisTypeMapper {
// ------------------------------time-------------------------
private static final String DORIS_DATE = "DATE";
+ private static final String DORIS_DATEV2 = "DATEV2";
private static final String DORIS_DATETIME = "DATETIME";
+ private static final String DORIS_DATETIMEV2 = "DATETIMEV2";
//------------------------------bool------------------------
private static final String DORIS_BOOLEAN = "BOOLEAN";
@@ -64,6 +70,10 @@ public class DorisTypeMapper {
case DORIS_BIGINT:
return DataTypes.BIGINT();
case DORIS_DECIMAL:
+ case DORIS_DECIMALV2:
+ case DORIS_DECIMAL32:
+ case DORIS_DECIMAL64:
+ case DORIS_DECIMAL128I:
return DataTypes.DECIMAL(precision, scale);
case DORIS_FLOAT:
return DataTypes.FLOAT();
@@ -77,8 +87,10 @@ public class DorisTypeMapper {
case DORIS_TEXT:
return DataTypes.STRING();
case DORIS_DATE:
+ case DORIS_DATEV2:
return DataTypes.DATE();
case DORIS_DATETIME:
+ case DORIS_DATETIMEV2:
return DataTypes.TIMESTAMP(0);
default:
throw new UnsupportedOperationException(
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/serialization/RowBatch.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
index e83300a..1cd9fdc 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/serialization/RowBatch.java
@@ -85,6 +85,7 @@ public class RowBatch {
private final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+ private final DateTimeFormatter dateTimeV2Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
public List<Row> getRowBatch() {
return rowBatch;
@@ -237,6 +238,9 @@ public class RowBatch {
break;
case "DECIMAL":
case "DECIMALV2":
+ case "DECIMAL32":
+ case "DECIMAL64":
+ case "DECIMAL128I":
Preconditions.checkArgument(mt.equals(Types.MinorType.DECIMAL),
typeMismatchMessage(currentType, mt));
DecimalVector decimalVector = (DecimalVector) curFieldVector;
@@ -250,6 +254,7 @@ public class RowBatch {
}
break;
case "DATE":
+ case "DATEV2":
Preconditions.checkArgument(mt.equals(Types.MinorType.VARCHAR),
typeMismatchMessage(currentType, mt));
VarCharVector date = (VarCharVector) curFieldVector;
@@ -277,6 +282,20 @@ public class RowBatch {
addValueToRow(rowIndex, parse);
}
break;
+ case "DATETIMEV2":
+ Preconditions.checkArgument(mt.equals(Types.MinorType.VARCHAR),
+ typeMismatchMessage(currentType, mt));
+ VarCharVector timeStampV2SecVector = (VarCharVector) curFieldVector;
+ for (int rowIndex = 0; rowIndex < rowCountInOneBatch; rowIndex++) {
+ if (timeStampV2SecVector.isNull(rowIndex)) {
+ addValueToRow(rowIndex, null);
+ continue;
+ }
+ String value = new String(timeStampV2SecVector.get(rowIndex));
+ LocalDateTime parse = LocalDateTime.parse(value, dateTimeV2Formatter);
+ addValueToRow(rowIndex, parse);
+ }
+ break;
case "LARGEINT":
case "CHAR":
case "VARCHAR":
diff --git a/flink-doris-connector/src/main/thrift/doris/Types.thrift b/flink-doris-connector/src/main/thrift/doris/Types.thrift
index 44ce606..a30afde 100644
--- a/flink-doris-connector/src/main/thrift/doris/Types.thrift
+++ b/flink-doris-connector/src/main/thrift/doris/Types.thrift
@@ -66,7 +66,7 @@ enum TPrimitiveType {
DATE,
DATETIME,
BINARY,
- DECIMAL,
+ DECIMAL_DEPRACTED, // not used now, only for place holder
// CHAR(n). Currently only supported in UDAs
CHAR,
LARGEINT,
@@ -79,7 +79,16 @@ enum TPrimitiveType {
MAP,
STRUCT,
STRING,
- ALL
+ ALL,
+ QUANTILE_STATE,
+ DATEV2,
+ DATETIMEV2,
+ TIMEV2,
+ DECIMAL32,
+ DECIMAL64,
+ DECIMAL128I,
+ JSONB,
+ UNSUPPORTED
}
enum TTypeNodeType {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org