This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 9b2d634 [INLONG-677] Support dt as built-in data time partition field
in inlong-sort (#519)
9b2d634 is described below
commit 9b2d634c5a5b67b8e5205d7ab3fbbcea294166b2
Author: chantccc <[email protected]>
AuthorDate: Wed Jul 7 19:09:11 2021 +0800
[INLONG-677] Support dt as built-in data time partition field in
inlong-sort (#519)
Co-authored-by: leobiaoliu <[email protected]>
---
.../inlong/sort/configuration/Constants.java | 5 ++++
.../transformation/FieldMappingTransformer.java | 12 +++++++---
.../FieldMappingTransformerTest.java | 27 ++++++++++++++++++++++
3 files changed, 41 insertions(+), 3 deletions(-)
diff --git
a/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
b/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
index dfc077a..4c6c67e 100644
---
a/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
+++
b/inlong-sort/common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
@@ -41,6 +41,11 @@ public class Constants {
public static final String SINK_UID = "sink_uid";
+ /**
+ * It uses dt as the built-in data time field name. It's a work-around
solution and should be replaced by later.
+ */
+ public static final String DATA_TIME_FIELD = "dt";
+
// ------------------------------------------------------------------------
// Common configs
// ------------------------------------------------------------------------
diff --git
a/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformer.java
b/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformer.java
index 1f5f2ff..e4485c2 100644
---
a/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformer.java
+++
b/inlong-sort/core/src/main/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformer.java
@@ -27,6 +27,8 @@ import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.sink.SinkInfo;
import org.apache.inlong.sort.protocol.source.SourceInfo;
+import static org.apache.inlong.sort.configuration.Constants.DATA_TIME_FIELD;
+
/**
* TODO, replace it with operator when it supports complex transformation.
*/
@@ -88,6 +90,7 @@ public class FieldMappingTransformer implements
DataFlowInfoListener {
for (int i = 0; i < sourceInfo.getFields().length; i++) {
final FieldInfo fieldInfo = sourceInfo.getFields()[i];
// use name to map field from source and sink
+ // skip time and attributes fields
sourceFieldIndices.put(fieldInfo.getName(), i +
SOURCE_FIELD_SKIP_STEP);
}
final SinkInfo sinkInfo = dataFlowInfo.getSinkInfo();
@@ -95,10 +98,13 @@ public class FieldMappingTransformer implements
DataFlowInfoListener {
for (int i = 0; i < sinkInfo.getFields().length; i++) {
final FieldInfo fieldInfo = sinkInfo.getFields()[i];
final Integer fieldIndex =
sourceFieldIndices.get(fieldInfo.getName());
- if (fieldIndex == null) {
- indices[i] = -1;
- } else {
+ if (fieldIndex != null) {
indices[i] = fieldIndex;
+ } else if (DATA_TIME_FIELD.equals(fieldInfo.getName())) {
+ // built-in data time field
+ indices[i] = 0;
+ } else {
+ indices[i] = -1;
}
}
}
diff --git
a/inlong-sort/core/src/test/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformerTest.java
b/inlong-sort/core/src/test/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformerTest.java
index ad76ac6..43aab2d 100644
---
a/inlong-sort/core/src/test/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformerTest.java
+++
b/inlong-sort/core/src/test/java/org/apache/inlong/sort/flink/transformation/FieldMappingTransformerTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.types.Row;
import org.apache.inlong.sort.flink.Record;
import org.apache.inlong.sort.formats.common.LongFormatInfo;
import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.formats.common.TimestampFormatInfo;
import org.apache.inlong.sort.protocol.DataFlowInfo;
import org.apache.inlong.sort.protocol.FieldInfo;
import org.apache.inlong.sort.protocol.sink.SinkInfo;
@@ -61,4 +62,30 @@ public class FieldMappingTransformerTest extends TestLogger {
assertEquals(1, sinkRecord.getRow().getArity());
assertEquals(9527L, sinkRecord.getRow().getField(0));
}
+
+ @Test
+ public void testTransformWithDt() throws Exception {
+ final FieldInfo fieldInfo = new FieldInfo("id", new LongFormatInfo());
+ final FieldInfo dtFieldInfo = new FieldInfo("dt", new
TimestampFormatInfo("MILLIS"));
+
+ final SourceInfo sourceInfo = new TestingSourceInfo(new
FieldInfo[]{fieldInfo});
+ final SinkInfo sinkInfo = new TestingSinkInfo(new
FieldInfo[]{fieldInfo, dtFieldInfo});
+ final long dataFlowId = 1L;
+ final DataFlowInfo dataFlowInfo = new DataFlowInfo(dataFlowId,
sourceInfo, sinkInfo);
+
+ final FieldMappingTransformer transformer = new
FieldMappingTransformer();
+ transformer.addDataFlow(dataFlowInfo);
+ // should be 3 fields (1 origin fields + time + attr)
+ final Row sourceRow = new Row(1 + SOURCE_FIELD_SKIP_STEP);
+ final long dt = System.currentTimeMillis();
+ sourceRow.setField(0, dt);
+ sourceRow.setField(1, "attr");
+ sourceRow.setField(2, 9527L);
+ final Record sourceRecord = new Record(dataFlowId, sourceRow);
+ final Record sinkRecord = transformer.transform(sourceRecord);
+ assertEquals(dataFlowId, sinkRecord.getDataflowId());
+ assertEquals(2, sinkRecord.getRow().getArity());
+ assertEquals(9527L, sinkRecord.getRow().getField(0));
+ assertEquals(dt, sinkRecord.getRow().getField(1));
+ }
}
\ No newline at end of file
