This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 87d8aef3ed [INLONG-8236][Sort] Iceberg supports dynamic switching
between append and upsert (#8238)
87d8aef3ed is described below
commit 87d8aef3ed4ab1180f74019247529f9ca55ab28c
Author: ChenLin <[email protected]>
AuthorDate: Wed Jul 26 11:19:50 2023 +0800
[INLONG-8236][Sort] Iceberg supports dynamic switching between append and
upsert (#8238)
---
.../org/apache/inlong/sort/base/Constants.java | 9 +++
.../inlong/sort/iceberg/IcebergTableSink.java | 4 ++
.../apache/inlong/sort/iceberg/sink/FlinkSink.java | 63 +++++++++++++++---
.../iceberg/sink/RowDataTaskWriterFactory.java | 18 +++++-
.../sink/multiple/DynamicSchemaHandleOperator.java | 7 ++
.../sink/multiple/IcebergMultipleStreamWriter.java | 24 +++++--
.../sink/multiple/IcebergSingleStreamWriter.java | 75 ++++++++++++++++++++--
.../iceberg/sink/multiple/RecordWithSchema.java | 1 +
8 files changed, 180 insertions(+), 21 deletions(-)
diff --git
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
index 5ac90a8077..4f31999985 100644
---
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
+++
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
@@ -165,6 +165,8 @@ public final class Constants {
public static final String GHOST_TAG = "/* gh-ost */";
+ public static final String META_INCREMENTAL = "meta.incremental";
+
public static final ConfigOption<String> INLONG_METRIC =
ConfigOptions.key("inlong.metric.labels")
.stringType()
@@ -241,6 +243,13 @@ public final class Constants {
.withDescription("The option 'sink.multiple.enable' "
+ "is used to determine whether to support
multiple sink writing, default is 'false'.");
+ public static final ConfigOption<Boolean> SWITCH_APPEND_UPSERT_ENABLE =
+ ConfigOptions.key("switch.append.upsert.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("The option 'switch.append.upsert.enable'
"
+ + "is used to switch between append and upsert,
default is 'false'.");
+
public static final ConfigOption<SchemaUpdateExceptionPolicy>
SINK_MULTIPLE_SCHEMA_UPDATE_POLICY =
ConfigOptions.key("sink.multiple.schema-update.policy")
.enumType(SchemaUpdateExceptionPolicy.class)
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java
index 22e8d39ba4..200acbb85a 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java
@@ -59,6 +59,7 @@ import static
org.apache.inlong.sort.base.Constants.SINK_MULTIPLE_TABLE_PATTERN;
import static
org.apache.inlong.sort.base.Constants.SINK_MULTIPLE_TYPE_MAP_COMPATIBLE_WITH_SPARK;
import static org.apache.inlong.sort.base.Constants.SINK_SCHEMA_CHANGE_ENABLE;
import static
org.apache.inlong.sort.base.Constants.SINK_SCHEMA_CHANGE_POLICIES;
+import static
org.apache.inlong.sort.base.Constants.SWITCH_APPEND_UPSERT_ENABLE;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_DISTRIBUTION_MODE;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_PARALLELISM;
@@ -131,7 +132,9 @@ public class IcebergTableSink implements DynamicTableSink,
SupportsPartitioning,
.appendMode(tableOptions.get(IGNORE_ALL_CHANGELOG))
.metric(tableOptions.get(INLONG_METRIC),
tableOptions.get(INLONG_AUDIT))
.catalogLoader(catalogLoader)
+ .tableSchema(tableSchema)
.multipleSink(tableOptions.get(SINK_MULTIPLE_ENABLE))
+
.switchAppendUpsertEnable(tableOptions.get(SWITCH_APPEND_UPSERT_ENABLE))
.multipleSinkOption(MultipleSinkOption.builder()
.withFormat(tableOptions.get(SINK_MULTIPLE_FORMAT))
.withSparkEngineEnable(tableOptions.get(SINK_MULTIPLE_TYPE_MAP_COMPATIBLE_WITH_SPARK))
@@ -163,6 +166,7 @@ public class IcebergTableSink implements DynamicTableSink,
SupportsPartitioning,
.action(actionsProvider)
.tableOptions(tableOptions)
.distributionMode(DistributionMode.fromName(tableOptions.get(WRITE_DISTRIBUTION_MODE)))
+
.switchAppendUpsertEnable(tableOptions.get(SWITCH_APPEND_UPSERT_ENABLE))
.append();
}
}
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java
index f36de37786..b3dd87b77c 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java
@@ -40,6 +40,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.RowData.FieldGetter;
@@ -62,7 +63,6 @@ import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.FlinkWriteConf;
import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.TableLoader;
-import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
import org.apache.iceberg.io.WriteResult;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
@@ -71,6 +71,7 @@ import
org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types.NestedField;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -85,7 +86,10 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.ROW;
import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
+import static org.apache.inlong.sort.base.Constants.META_INCREMENTAL;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_MINI_BATCH_BUFFER_TYPE;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_MINI_BATCH_ENABLE;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_MINI_BATCH_PRE_AGG_ENABLE;
@@ -184,6 +188,7 @@ public class FlinkSink {
private ReadableConfig tableOptions = new Configuration();
private boolean enableSchemaChange;
private String schemaChangePolicies;
+ private boolean switchAppendUpsertEnable = false;
private Builder() {
}
@@ -257,6 +262,11 @@ public class FlinkSink {
return this;
}
+ public Builder switchAppendUpsertEnable(boolean
switchAppendUpsertEnable) {
+ this.switchAppendUpsertEnable = switchAppendUpsertEnable;
+ return this;
+ }
+
public Builder multipleSinkOption(MultipleSinkOption
multipleSinkOption) {
this.multipleSinkOption = multipleSinkOption;
return this;
@@ -664,7 +674,8 @@ public class FlinkSink {
IcebergProcessOperator<RowData, WriteResult> streamWriter =
createStreamWriter(
table, flinkRowType, equalityFieldIds, flinkWriteConf,
appendMode, inlongMetric,
- auditHostAndPorts, dirtyOptions, dirtySink,
tableOptions.get(WRITE_MINI_BATCH_ENABLE));
+ auditHostAndPorts, dirtyOptions, dirtySink,
tableOptions.get(WRITE_MINI_BATCH_ENABLE), tableSchema,
+ switchAppendUpsertEnable);
int parallelism = writeParallelism == null ?
input.getParallelism() : writeParallelism;
SingleOutputStreamOperator<WriteResult> writerStream =
inputWithMiniBatch
@@ -691,11 +702,13 @@ public class FlinkSink {
TypeInformation.of(RecordWithSchema.class),
routeOperator)
.setParallelism(parallelism);
-
+ RowType tableSchemaRowType = (RowType)
tableSchema.toRowDataType().getLogicalType();
+ int metaFieldIndex = getMetaFieldIndex(tableSchema);
IcebergProcessOperator streamWriter =
new IcebergProcessOperator(new IcebergMultipleStreamWriter(
appendMode, catalogLoader, inlongMetric,
auditHostAndPorts,
- multipleSinkOption, dirtyOptions, dirtySink));
+ multipleSinkOption, dirtyOptions, dirtySink,
tableSchemaRowType, metaFieldIndex,
+ switchAppendUpsertEnable));
SingleOutputStreamOperator<MultipleWriteResult> writerStream =
routeStream
.transform(operatorName(ICEBERG_MULTIPLE_STREAM_WRITER_NAME),
TypeInformation.of(IcebergProcessOperator.class),
@@ -778,23 +791,50 @@ public class FlinkSink {
}
+ static DataType filterOutMetaField(TableSchema requestedSchema) {
+ DataTypes.Field[] fields = requestedSchema.getTableColumns().stream()
+ .filter(column -> !META_INCREMENTAL.equals(column.getName()))
+ .map(column -> FIELD(column.getName(), column.getType()))
+ .toArray(DataTypes.Field[]::new);
+ return ROW(fields).notNull();
+ }
+
static RowType toFlinkRowType(Schema schema, TableSchema requestedSchema) {
if (requestedSchema != null) {
// Convert the flink schema to iceberg schema firstly, then
reassign ids to match the existing iceberg
// schema.
- Schema writeSchema =
TypeUtil.reassignIds(FlinkSchemaUtil.convert(requestedSchema), schema);
+ List<NestedField> filteredFields =
FlinkSchemaUtil.convert(requestedSchema)
+ .columns()
+ .stream()
+ .filter(nestedField ->
!META_INCREMENTAL.equals(nestedField.name()))
+ .collect(Collectors.toList());
+ Schema writeSchema = TypeUtil.reassignIds(new
Schema(filteredFields), schema);
TypeUtil.validateWriteSchema(schema, writeSchema, true, true);
// We use this flink schema to read values from RowData. The
flink's TINYINT and SMALLINT will be promoted
// to iceberg INTEGER, that means if we use iceberg's table schema
to read TINYINT (backend by 1 'byte'),
// we will read 4 bytes rather than 1 byte, it will mess up the
byte array in BinaryRowData. So here we must
// use flink schema.
- return (RowType) requestedSchema.toRowDataType().getLogicalType();
+ return (RowType)
filterOutMetaField(requestedSchema).getLogicalType();
} else {
return FlinkSchemaUtil.convert(schema);
}
}
+ static int getMetaFieldIndex(TableSchema tableSchema) {
+ RowType rowType = (RowType)
tableSchema.toRowDataType().getLogicalType();
+ List<RowType.RowField> fields = rowType.getFields();
+ int metaFieldIndex = -1;
+ for (int i = 0; i < fields.size(); i++) {
+ RowType.RowField rowField = fields.get(i);
+ if (META_INCREMENTAL.equals(rowField.getName())) {
+ metaFieldIndex = i;
+ break;
+ }
+ }
+ return metaFieldIndex;
+ }
+
static IcebergProcessOperator<RowData, WriteResult>
createStreamWriter(Table table,
RowType flinkRowType,
List<Integer> equalityFieldIds,
@@ -804,12 +844,14 @@ public class FlinkSink {
String auditHostAndPorts,
DirtyOptions dirtyOptions,
@Nullable DirtySink<Object> dirtySink,
- boolean miniBatchMode) {
+ boolean miniBatchMode,
+ TableSchema tableSchema,
+ boolean switchAppendUpsertEnable) {
// flink A, iceberg a
Preconditions.checkArgument(table != null, "Iceberg table should't be
null");
Table serializableTable = SerializableTable.copyOf(table);
- TaskWriterFactory<RowData> taskWriterFactory =
+ RowDataTaskWriterFactory taskWriterFactory =
new RowDataTaskWriterFactory(
serializableTable,
serializableTable.schema(),
@@ -821,9 +863,12 @@ public class FlinkSink {
appendMode,
miniBatchMode);
+ RowType tableSchemaRowType = (RowType)
tableSchema.toRowDataType().getLogicalType();
+ int metaFieldIndex = getMetaFieldIndex(tableSchema);
return new IcebergProcessOperator<>(new IcebergSingleStreamWriter<>(
table.name(), taskWriterFactory, inlongMetric,
auditHostAndPorts,
- null, dirtyOptions, dirtySink, false));
+ flinkRowType, dirtyOptions, dirtySink, false,
+ tableSchemaRowType, metaFieldIndex, switchAppendUpsertEnable));
}
}
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java
index 44341feffb..b682fc1f7f 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/RowDataTaskWriterFactory.java
@@ -54,8 +54,8 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
private final long targetFileSizeBytes;
private final FileFormat format;
private final List<Integer> equalityFieldIds;
- private final boolean upsert;
- private final boolean appendMode;
+ private boolean upsert;
+ private boolean appendMode;
private final boolean miniBatchMode;
private final FileAppenderFactory<RowData> appenderFactory;
@@ -97,6 +97,20 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
}
}
+ public void switchToUpsert() {
+ this.appendMode = false;
+ this.upsert = true;
+ }
+
+ public void switchToAppend() {
+ this.appendMode = true;
+ this.upsert = false;
+ }
+
+ public boolean isAppendMode() {
+ return equalityFieldIds == null || equalityFieldIds.isEmpty() ||
appendMode;
+ }
+
@Override
public void initialize(int taskId, int attemptId) {
this.outputFileFactory = OutputFileFactory.builderFor(table, taskId,
attemptId).build();
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
index 1b12639d57..94caad6b52 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/DynamicSchemaHandleOperator.java
@@ -77,6 +77,7 @@ import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -125,6 +126,7 @@ public class DynamicSchemaHandleOperator extends
AbstractStreamOperator<RecordWi
private SchemaChangeHelper schemaChangeHelper;
private String schemaChangePolicies;
private boolean enableSchemaChange;
+ private final String INCREMENTAL = "incremental";
public DynamicSchemaHandleOperator(CatalogLoader catalogLoader,
MultipleSinkOption multipleSinkOption,
@@ -423,6 +425,11 @@ public class DynamicSchemaHandleOperator extends
AbstractStreamOperator<RecordWi
recordWithSchema.setDirty(isDirty.get());
recordWithSchema.setRowCount(rowCount.get());
recordWithSchema.setRowSize(rowSize.get());
+ JsonNode originalData = recordWithSchema.getOriginalData();
+ boolean incremental =
Optional.ofNullable(originalData.get(INCREMENTAL))
+ .map(node -> node.asBoolean())
+ .orElse(false);
+ recordWithSchema.setIncremental(incremental);
output.collect(new StreamRecord<>(recordWithSchema));
} else {
if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE ==
multipleSinkOption
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java
index 4652643bec..3cd909ef5b 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java
@@ -50,7 +50,6 @@ import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.CatalogLoader;
import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.PropertyUtil;
import org.slf4j.Logger;
@@ -110,6 +109,9 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
private transient MetricState metricState;
private transient ListState<MetricState> metricStateListState;
private transient RuntimeContext runtimeContext;
+ private final RowType tableSchemaRowType;
+ private final int metaFieldIndex;
+ private final boolean switchAppendUpsertEnable;
public IcebergMultipleStreamWriter(
boolean appendMode,
@@ -118,7 +120,10 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
String auditHostAndPorts,
MultipleSinkOption multipleSinkOption,
DirtyOptions dirtyOptions,
- @Nullable DirtySink<Object> dirtySink) {
+ @Nullable DirtySink<Object> dirtySink,
+ RowType tableSchemaRowType,
+ int metaFieldIndex,
+ boolean switchAppendUpsertEnable) {
this.appendMode = appendMode;
this.catalogLoader = catalogLoader;
this.inlongMetric = inlongMetric;
@@ -126,6 +131,9 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
this.multipleSinkOption = multipleSinkOption;
this.dirtyOptions = dirtyOptions;
this.dirtySink = dirtySink;
+ this.tableSchemaRowType = tableSchemaRowType;
+ this.metaFieldIndex = metaFieldIndex;
+ this.switchAppendUpsertEnable = switchAppendUpsertEnable;
}
@Override
@@ -213,7 +221,7 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
.collect(Collectors.toList());
}
RowType flinkRowType =
FlinkSchemaUtil.convert(recordWithSchema.getSchema());
- TaskWriterFactory<RowData> taskWriterFactory = new
RowDataTaskWriterFactory(
+ RowDataTaskWriterFactory taskWriterFactory = new
RowDataTaskWriterFactory(
table,
recordWithSchema.getSchema(),
flinkRowType,
@@ -232,7 +240,8 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
.append(Constants.TABLE_NAME).append("=").append(tableId.name());
IcebergSingleStreamWriter<RowData> writer = new
IcebergSingleStreamWriter<>(
tableId.toString(), taskWriterFactory,
subWriterInlongMetric.toString(),
- auditHostAndPorts, flinkRowType, dirtyOptions,
dirtySink, true);
+ auditHostAndPorts, flinkRowType, dirtyOptions,
dirtySink, true,
+ tableSchemaRowType, metaFieldIndex,
switchAppendUpsertEnable);
writer.setup(getRuntimeContext(),
new CallbackCollector<>(
writeResult -> collector.collect(new
MultipleWriteResult(tableId, writeResult))),
@@ -263,6 +272,13 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
long size = CalculateObjectSizeUtils.getDataSize(data);
try {
+ if (switchAppendUpsertEnable) {
+ if (recordWithSchema.isIncremental()) {
+ multipleWriters.get(tableId).switchToUpsert();
+ } else {
+ multipleWriters.get(tableId).switchToAppend();
+ }
+ }
multipleWriters.get(tableId).processElement(data);
} catch (Exception e) {
LOG.error(String.format("write error, raw data: %s",
data), e);
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java
index 13eaefac24..7d491be680 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java
@@ -25,6 +25,7 @@ import
org.apache.inlong.sort.base.metric.MetricOption.RegisteredMetric;
import org.apache.inlong.sort.base.metric.MetricState;
import org.apache.inlong.sort.base.metric.SinkMetricData;
import org.apache.inlong.sort.base.util.MetricStateUtils;
+import org.apache.inlong.sort.iceberg.sink.RowDataTaskWriterFactory;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
@@ -34,6 +35,8 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.io.TaskWriter;
@@ -45,6 +48,8 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import static org.apache.inlong.sort.base.Constants.DIRTY_BYTES_OUT;
import static org.apache.inlong.sort.base.Constants.DIRTY_RECORDS_OUT;
@@ -64,9 +69,10 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
private final String fullTableName;
private final String inlongMetric;
private final String auditHostAndPorts;
- private TaskWriterFactory<T> taskWriterFactory;
+ private RowDataTaskWriterFactory taskWriterFactory;
+
+ private transient TaskWriter<RowData> writer;
- private transient TaskWriter<T> writer;
private transient int subTaskId;
private transient int attemptId;
private @Nullable transient SinkMetricData metricData;
@@ -76,16 +82,23 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
private final DirtyOptions dirtyOptions;
private @Nullable final DirtySink<Object> dirtySink;
private boolean multipleSink;
+ private final RowType tableSchemaRowType;
+ private final int metaFieldIndex;
+ private final List<WriteResult> cachedWriteResults;
+ private final boolean switchAppendUpsertEnable;
public IcebergSingleStreamWriter(
String fullTableName,
- TaskWriterFactory<T> taskWriterFactory,
+ RowDataTaskWriterFactory taskWriterFactory,
String inlongMetric,
String auditHostAndPorts,
@Nullable RowType flinkRowType,
DirtyOptions dirtyOptions,
@Nullable DirtySink<Object> dirtySink,
- boolean multipleSink) {
+ boolean multipleSink,
+ RowType tableSchemaRowType,
+ int metaFieldIndex,
+ boolean switchAppendUpsertEnable) {
this.fullTableName = fullTableName;
this.taskWriterFactory = taskWriterFactory;
this.inlongMetric = inlongMetric;
@@ -94,6 +107,10 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
this.dirtyOptions = dirtyOptions;
this.dirtySink = dirtySink;
this.multipleSink = multipleSink;
+ this.tableSchemaRowType = tableSchemaRowType;
+ this.metaFieldIndex = metaFieldIndex;
+ this.cachedWriteResults = new ArrayList<>();
+ this.switchAppendUpsertEnable = switchAppendUpsertEnable;
}
public RowType getFlinkRowType() {
@@ -137,15 +154,61 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ // submit the cached write results
+ cachedWriteResults.forEach(writeResult -> emit(writeResult));
+
// close all open files and emit files to downstream committer operator
emit(writer.complete());
this.writer = taskWriterFactory.create();
}
+ private RowData removeField(RowData rowData, int fieldIndex, RowType
rowType) {
+ GenericRowData newRowData = new GenericRowData(rowType.getFieldCount()
- 1);
+
+ for (int i = 0, j = 0; i < rowType.getFieldCount(); i++) {
+ if (i != fieldIndex) {
+ newRowData.setField(j++, rowData.getRawValue(i));
+ }
+ }
+
+ return newRowData;
+ }
+
+ private void cacheWriteResultAndRecreateWriter() throws IOException {
+ // close all open file and cache writeResult
+ cachedWriteResults.add(writer.complete());
+ this.writer = taskWriterFactory.create();
+ }
+
+ public void switchToUpsert() throws Exception {
+ if (taskWriterFactory.isAppendMode()) {
+ taskWriterFactory.switchToUpsert();
+ cacheWriteResultAndRecreateWriter();
+ }
+ }
+
+ public void switchToAppend() throws Exception {
+ if (taskWriterFactory.isAppendMode())
+ return;
+ taskWriterFactory.switchToAppend();
+ cacheWriteResultAndRecreateWriter();
+ }
+
@Override
public void processElement(T value) throws Exception {
try {
- writer.write(value);
+ if (!switchAppendUpsertEnable || multipleSink || metaFieldIndex ==
-1) {
+ writer.write((RowData) value);
+ return;
+ }
+
+ RowData rowData = (RowData) value;
+ if (rowData.getBoolean(metaFieldIndex)) {
+ switchToUpsert();
+ } else {
+ switchToAppend();
+ }
+ writer.write(removeField(rowData, metaFieldIndex,
tableSchemaRowType));
} catch (Exception e) {
if (multipleSink) {
throw e;
@@ -232,7 +295,7 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
public void schemaEvolution(TaskWriterFactory<T> schema) throws
IOException {
emit(writer.complete());
- taskWriterFactory = schema;
+ taskWriterFactory = (RowDataTaskWriterFactory) schema;
taskWriterFactory.initialize(subTaskId, attemptId);
writer = taskWriterFactory.create();
}
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/RecordWithSchema.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/RecordWithSchema.java
index 1806ade47a..c929ab3742 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/RecordWithSchema.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/RecordWithSchema.java
@@ -61,6 +61,7 @@ public class RecordWithSchema {
private boolean isDirty;
private long rowCount;
private long rowSize;
+ private boolean incremental;
private transient JsonNode originalData;
