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 5b0f5cb7c6 [INLONG-8596][Sort] Iceberg supports dynamic switching
between append and upsert (#8599)
5b0f5cb7c6 is described below
commit 5b0f5cb7c6a4ba1ce2532b5b6557ea597d849e9b
Author: Sting <[email protected]>
AuthorDate: Fri Jul 28 12:50:11 2023 +0800
[INLONG-8596][Sort] Iceberg supports dynamic switching between append and
upsert (#8599)
---
.../org/apache/inlong/sort/base/Constants.java | 5 +-
.../inlong/sort/iceberg/IcebergTableSink.java | 3 +-
.../iceberg/schema/IcebergModeSwitchHelper.java | 101 +++++++++++++++++++++
.../sort/iceberg/schema/RowDataConverter.java | 47 ++++++++++
.../apache/inlong/sort/iceberg/sink/FlinkSink.java | 48 +++-------
.../iceberg/sink/RowDataTaskWriterFactory.java | 26 ++++--
.../sink/multiple/DynamicSchemaHandleOperator.java | 6 +-
.../sink/multiple/IcebergMultipleStreamWriter.java | 19 ++--
.../sink/multiple/IcebergSingleFileCommiter.java | 1 +
.../sink/multiple/IcebergSingleStreamWriter.java | 98 ++++++++++++--------
10 files changed, 255 insertions(+), 99 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 4f31999985..7a42c892d0 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,7 +165,7 @@ public final class Constants {
public static final String GHOST_TAG = "/* gh-ost */";
- public static final String META_INCREMENTAL = "meta.incremental";
+ public static final String META_INCREMENTAL = "incremental_inlong";
public static final ConfigOption<String> INLONG_METRIC =
ConfigOptions.key("inlong.metric.labels")
@@ -248,7 +248,8 @@ public final class Constants {
.booleanType()
.defaultValue(false)
.withDescription("The option 'switch.append.upsert.enable'
"
- + "is used to switch between append and upsert,
default is 'false'.");
+ + "is used when sink connector switch between
append and upsert mode, "
+ + "default is 'false'.");
public static final ConfigOption<SchemaUpdateExceptionPolicy>
SINK_MULTIPLE_SCHEMA_UPDATE_POLICY =
ConfigOptions.key("sink.multiple.schema-update.policy")
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 200acbb85a..f57b2c62e0 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
@@ -121,11 +121,12 @@ public class IcebergTableSink implements
DynamicTableSink, SupportsPartitioning,
List<String> equalityColumns = tableSchema.getPrimaryKey()
.map(UniqueConstraint::getColumns)
.orElseGet(ImmutableList::of);
-
+ LOG.info("iceberg sink running with equality columns {}",
equalityColumns);
final ReadableConfig tableOptions =
Configuration.fromMap(catalogTable.getOptions());
boolean multipleSink = tableOptions.get(SINK_MULTIPLE_ENABLE);
boolean schemaChange = tableOptions.get(SINK_SCHEMA_CHANGE_ENABLE);
String schemaChangePolicies =
tableOptions.getOptional(SINK_SCHEMA_CHANGE_POLICIES).orElse(null);
+ LOG.info("iceberg sink running with policy {}",
tableOptions.get(SINK_MULTIPLE_SCHEMA_UPDATE_POLICY));
if (multipleSink) {
return (DataStreamSinkProvider) dataStream ->
FlinkSink.forRowData(dataStream)
.overwrite(overwrite)
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/IcebergModeSwitchHelper.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/IcebergModeSwitchHelper.java
new file mode 100644
index 0000000000..88a743f6c4
--- /dev/null
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/IcebergModeSwitchHelper.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.iceberg.schema;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.RowType.RowField;
+
+import java.util.List;
+
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.ROW;
+
+/**
+ * this class helps to manipulate iceberg table schema
+ * when SWITCH_APPEND_UPSERT_ENABLE equals to true
+ */
+public class IcebergModeSwitchHelper {
+
+ private final RowType tableSchemaRowType;
+ private final RowDataConverter rowDataConverter;
+ private final int incrementalFieldIndex;
+ public static final String META_INCREMENTAL = "incremental_inlong";
+ public static final int DEFAULT_META_INDEX = -1;
+
+ public IcebergModeSwitchHelper(RowType tableSchemaRowType, int
incrementalFieldIndex) {
+ this.tableSchemaRowType = tableSchemaRowType;
+ this.rowDataConverter = new
RowDataConverter(tableSchemaRowType.getChildren());
+ this.incrementalFieldIndex = incrementalFieldIndex;
+ }
+
+ /**
+ * remove incremental field from rowData
+ * @param rowData input row data
+ * @return row data without incremental field
+ */
+ public RowData removeIncrementalField(RowData rowData) {
+ if (incrementalFieldIndex == DEFAULT_META_INDEX) {
+ return rowData;
+ }
+ GenericRowData newRowData = new
GenericRowData(tableSchemaRowType.getFieldCount() - 1);
+ for (int i = 0, j = 0; i < tableSchemaRowType.getFieldCount(); i++) {
+ if (i != incrementalFieldIndex) {
+ newRowData.setField(j++, rowDataConverter.get(rowData, i));
+ }
+ }
+ return newRowData;
+ }
+
+ /**
+ * remove incremental field from table schema
+ * @param requestedSchema input table schema
+ * @return table schema without incremental field
+ */
+ public 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();
+ }
+
+ /**
+ * get incremental field index
+ * @param tableSchema input table schema
+ * @return incremental field index
+ */
+ public static int getMetaFieldIndex(TableSchema tableSchema) {
+ RowType rowType = (RowType)
tableSchema.toRowDataType().getLogicalType();
+ List<RowField> fields = rowType.getFields();
+ int metaFieldIndex = DEFAULT_META_INDEX;
+ 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;
+ }
+
+}
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/RowDataConverter.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/RowDataConverter.java
new file mode 100644
index 0000000000..a3de8636a0
--- /dev/null
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/schema/RowDataConverter.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sort.iceberg.schema;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.List;
+
+import static
org.apache.inlong.sort.base.Constants.SWITCH_APPEND_UPSERT_ENABLE;
+
+/**
+ * used in iceberg running in {@link SWITCH_APPEND_UPSERT_ENABLE}
+ */
+public class RowDataConverter {
+
+ private final RowData.FieldGetter[] fieldGetter;
+
+ public RowDataConverter(List<LogicalType> types) {
+ this.fieldGetter = new RowData.FieldGetter[types.size()];
+
+ for (int i = 0; i < types.size(); ++i) {
+ this.fieldGetter[i] = RowData.createFieldGetter(types.get(i), i);
+ }
+
+ }
+
+ public Object get(RowData struct, int index) {
+ return this.fieldGetter[index].getFieldOrNull(struct);
+ }
+
+}
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 b3dd87b77c..c023b99530 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,7 +40,6 @@ 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;
@@ -79,6 +78,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -86,14 +86,14 @@ 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;
import static
org.apache.inlong.sort.iceberg.FlinkDynamicTableFactory.WRITE_RATE_LIMIT;
+import static
org.apache.inlong.sort.iceberg.schema.IcebergModeSwitchHelper.filterOutMetaField;
+import static
org.apache.inlong.sort.iceberg.schema.IcebergModeSwitchHelper.getMetaFieldIndex;
/**
* Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
@@ -536,6 +536,7 @@ public class FlinkSink {
}
equalityFieldIds = Lists.newArrayList(equalityFieldSet);
}
+ LOG.info("equalityFieldIds in iceberg: {}", equalityFieldIds);
return equalityFieldIds;
}
@@ -652,7 +653,9 @@ public class FlinkSink {
// Only if not appendMode, upsert can be valid.
boolean upsertMode = flinkWriteConf.upsertMode() && !appendMode;
+ LOG.info("The iceberg sink is using {} mode.", upsertMode ?
"upsert" : "append");
// Validate the equality fields and partition fields if we enable
the upsert mode.
+
if (upsertMode) {
Preconditions.checkState(!flinkWriteConf.overwriteMode(),
"OVERWRITE mode shouldn't be enable when configuring
to use UPSERT data stream.");
@@ -670,12 +673,12 @@ public class FlinkSink {
// Add rate limit if necessary
DataStream<RowData> inputWithRateLimit =
appendWithRateLimit(input);
DataStream<RowData> inputWithMiniBatch = appendWithMiniBatchGroup(
- inputWithRateLimit, flinkRowType,
equalityFieldIds.stream().collect(Collectors.toSet()));
+ inputWithRateLimit, flinkRowType, new
HashSet<>(equalityFieldIds));
IcebergProcessOperator<RowData, WriteResult> streamWriter =
createStreamWriter(
table, flinkRowType, equalityFieldIds, flinkWriteConf,
appendMode, inlongMetric,
- auditHostAndPorts, dirtyOptions, dirtySink,
tableOptions.get(WRITE_MINI_BATCH_ENABLE), tableSchema,
- switchAppendUpsertEnable);
+ auditHostAndPorts, dirtyOptions, dirtySink, tableSchema,
+ switchAppendUpsertEnable,
tableOptions.get(WRITE_MINI_BATCH_ENABLE));
int parallelism = writeParallelism == null ?
input.getParallelism() : writeParallelism;
SingleOutputStreamOperator<WriteResult> writerStream =
inputWithMiniBatch
@@ -703,11 +706,11 @@ public class FlinkSink {
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,
tableSchemaRowType, metaFieldIndex,
+ multipleSinkOption, dirtyOptions, dirtySink,
tableSchemaRowType,
+ getMetaFieldIndex(tableSchema),
switchAppendUpsertEnable));
SingleOutputStreamOperator<MultipleWriteResult> writerStream =
routeStream
.transform(operatorName(ICEBERG_MULTIPLE_STREAM_WRITER_NAME),
@@ -791,14 +794,6 @@ 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
@@ -821,20 +816,6 @@ public class FlinkSink {
}
}
- 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,
@@ -844,9 +825,9 @@ public class FlinkSink {
String auditHostAndPorts,
DirtyOptions dirtyOptions,
@Nullable DirtySink<Object> dirtySink,
- boolean miniBatchMode,
TableSchema tableSchema,
- boolean switchAppendUpsertEnable) {
+ boolean switchAppendUpsertEnable,
+ boolean miniBatchMode) {
// flink A, iceberg a
Preconditions.checkArgument(table != null, "Iceberg table should't be
null");
@@ -864,11 +845,10 @@ public class FlinkSink {
miniBatchMode);
RowType tableSchemaRowType = (RowType)
tableSchema.toRowDataType().getLogicalType();
- int metaFieldIndex = getMetaFieldIndex(tableSchema);
return new IcebergProcessOperator<>(new IcebergSingleStreamWriter<>(
table.name(), taskWriterFactory, inlongMetric,
auditHostAndPorts,
flinkRowType, dirtyOptions, dirtySink, false,
- tableSchemaRowType, metaFieldIndex, switchAppendUpsertEnable));
+ tableSchemaRowType, getMetaFieldIndex(tableSchema),
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 b682fc1f7f..5590e9bcc3 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
@@ -37,6 +37,8 @@ import
org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.ArrayUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
@@ -46,6 +48,8 @@ import java.util.List;
*/
public class RowDataTaskWriterFactory implements TaskWriterFactory<RowData> {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(RowDataTaskWriterFactory.class);
+
private final Table table;
private final Schema schema;
private final RowType flinkSchema;
@@ -56,9 +60,8 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
private final List<Integer> equalityFieldIds;
private boolean upsert;
private boolean appendMode;
+ private FileAppenderFactory<RowData> appenderFactory;
private final boolean miniBatchMode;
- private final FileAppenderFactory<RowData> appenderFactory;
-
private transient OutputFileFactory outputFileFactory;
public RowDataTaskWriterFactory(Table table,
@@ -80,19 +83,24 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
this.equalityFieldIds = equalityFieldIds;
this.upsert = upsert;
this.appendMode = appendMode;
+ this.appenderFactory = createRowDataFileAppenderFactory(table,
flinkSchema,
+ equalityFieldIds, upsert, appendMode);
this.miniBatchMode = miniBatchMode;
+ }
+ private FileAppenderFactory<RowData>
createRowDataFileAppenderFactory(Table table,
+ RowType flinkSchema, List<Integer> equalityFieldIds, boolean
upsert, boolean appendMode) {
if (equalityFieldIds == null || equalityFieldIds.isEmpty() ||
appendMode) {
- this.appenderFactory = new FlinkAppenderFactory(schema,
flinkSchema, table.properties(), spec);
+ return new FlinkAppenderFactory(schema, flinkSchema,
table.properties(), spec);
} else if (upsert) {
// In upsert mode, only the new row is emitted using INSERT row
kind. Therefore, any column of the inserted
// row may differ from the deleted row other than the primary key
fields, and the delete file must contain
// values that are correct for the deleted row. Therefore, only
write the equality delete fields.
- this.appenderFactory = new FlinkAppenderFactory(schema,
flinkSchema, table.properties(), spec,
+ return new FlinkAppenderFactory(schema, flinkSchema,
table.properties(), spec,
ArrayUtil.toIntArray(equalityFieldIds),
TypeUtil.select(schema,
Sets.newHashSet(equalityFieldIds)), null);
} else {
- this.appenderFactory = new FlinkAppenderFactory(schema,
flinkSchema, table.properties(), spec,
+ return new FlinkAppenderFactory(schema, flinkSchema,
table.properties(), spec,
ArrayUtil.toIntArray(equalityFieldIds), schema, null);
}
}
@@ -107,8 +115,8 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
this.upsert = false;
}
- public boolean isAppendMode() {
- return equalityFieldIds == null || equalityFieldIds.isEmpty() ||
appendMode;
+ public boolean isUpsert() {
+ return upsert;
}
@Override
@@ -124,9 +132,11 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
if (equalityFieldIds == null || equalityFieldIds.isEmpty() ||
appendMode) {
// Initialize a task writer to write INSERT only.
if (spec.isUnpartitioned()) {
+ LOGGER.info("Create an unPartitioned append writer for table
{}.", table.name());
return new UnpartitionedWriter<>(
spec, format, appenderFactory, outputFileFactory, io,
targetFileSizeBytes);
} else {
+ LOGGER.info("Create a partitioned append writer for table
{}.", table.name());
if (miniBatchMode) {
return new RowDataGroupedPartitionedFanoutWriter(spec,
format, appenderFactory, outputFileFactory,
io, targetFileSizeBytes, schema, flinkSchema);
@@ -138,9 +148,11 @@ public class RowDataTaskWriterFactory implements
TaskWriterFactory<RowData> {
} else {
// Initialize a task writer to write both INSERT and equality
DELETE.
if (spec.isUnpartitioned()) {
+ LOGGER.info("Create an unPartitioned upsert delta writer for
table {}.", table.name());
return new UnpartitionedDeltaWriter(spec, format,
appenderFactory, outputFileFactory, io,
targetFileSizeBytes, schema, flinkSchema,
equalityFieldIds, upsert);
} else {
+ LOGGER.info("Create a partitioned upsert delta writer for
table {}.", table.name());
if (miniBatchMode) {
return new GroupedPartitionedDeltaWriter(spec, format,
appenderFactory, outputFileFactory, io,
targetFileSizeBytes, schema, flinkSchema,
equalityFieldIds, upsert);
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 94caad6b52..2591c4fd59 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
@@ -426,10 +426,8 @@ public class DynamicSchemaHandleOperator extends
AbstractStreamOperator<RecordWi
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);
+
recordWithSchema.setIncremental(Optional.ofNullable(originalData.get(INCREMENTAL))
+ .map(JsonNode::asBoolean).orElse(false));
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 3cd909ef5b..88a2e27474 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
@@ -233,13 +233,12 @@ public class IcebergMultipleStreamWriter extends
IcebergProcessFunction<RecordWi
false);
if (multipleWriters.get(tableId) == null) {
- StringBuilder subWriterInlongMetric = new
StringBuilder(inlongMetric);
- subWriterInlongMetric.append(DELIMITER)
-
.append(Constants.DATABASE_NAME).append("=").append(tableId.namespace().toString())
- .append(DELIMITER)
-
.append(Constants.TABLE_NAME).append("=").append(tableId.name());
+ String subWriterInlongMetric = inlongMetric + DELIMITER
+ + Constants.DATABASE_NAME + "=" +
tableId.namespace().toString()
+ + DELIMITER
+ + Constants.TABLE_NAME + "=" + tableId.name();
IcebergSingleStreamWriter<RowData> writer = new
IcebergSingleStreamWriter<>(
- tableId.toString(), taskWriterFactory,
subWriterInlongMetric.toString(),
+ tableId.toString(), taskWriterFactory,
subWriterInlongMetric,
auditHostAndPorts, flinkRowType, dirtyOptions,
dirtySink, true,
tableSchemaRowType, metaFieldIndex,
switchAppendUpsertEnable);
writer.setup(getRuntimeContext(),
@@ -272,12 +271,8 @@ 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();
- }
+ if (switchAppendUpsertEnable &&
recordWithSchema.isIncremental()) {
+ multipleWriters.get(tableId).switchToUpsert();
}
multipleWriters.get(tableId).processElement(data);
} catch (Exception 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/IcebergSingleFileCommiter.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
index c8dee94171..074feec342 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
@@ -317,6 +317,7 @@ public class IcebergSingleFileCommiter extends
IcebergProcessFunction<WriteResul
continuousEmptyCheckpoints = 0;
}
// remove already committed snapshot manifest info
+
pendingMap.keySet().forEach(deltaManifestsMap::remove);
pendingMap.clear();
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 7d491be680..b197f42b31 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.schema.IcebergModeSwitchHelper;
import org.apache.inlong.sort.iceberg.sink.RowDataTaskWriterFactory;
import org.apache.flink.api.common.state.ListState;
@@ -35,7 +36,6 @@ 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;
@@ -56,6 +56,7 @@ import static
org.apache.inlong.sort.base.Constants.DIRTY_RECORDS_OUT;
import static org.apache.inlong.sort.base.Constants.INLONG_METRIC_STATE_NAME;
import static org.apache.inlong.sort.base.Constants.NUM_BYTES_OUT;
import static org.apache.inlong.sort.base.Constants.NUM_RECORDS_OUT;
+import static
org.apache.inlong.sort.iceberg.schema.IcebergModeSwitchHelper.DEFAULT_META_INDEX;
public class IcebergSingleStreamWriter<T> extends IcebergProcessFunction<T,
WriteResult>
implements
@@ -83,9 +84,10 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
private @Nullable final DirtySink<Object> dirtySink;
private boolean multipleSink;
private final RowType tableSchemaRowType;
- private final int metaFieldIndex;
+ private final int incrementalFieldIndex;
private final List<WriteResult> cachedWriteResults;
private final boolean switchAppendUpsertEnable;
+ private IcebergModeSwitchHelper switchHelper;
public IcebergSingleStreamWriter(
String fullTableName,
@@ -97,7 +99,7 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
@Nullable DirtySink<Object> dirtySink,
boolean multipleSink,
RowType tableSchemaRowType,
- int metaFieldIndex,
+ int incrementalFieldIndex,
boolean switchAppendUpsertEnable) {
this.fullTableName = fullTableName;
this.taskWriterFactory = taskWriterFactory;
@@ -108,7 +110,7 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
this.dirtySink = dirtySink;
this.multipleSink = multipleSink;
this.tableSchemaRowType = tableSchemaRowType;
- this.metaFieldIndex = metaFieldIndex;
+ this.incrementalFieldIndex = incrementalFieldIndex;
this.cachedWriteResults = new ArrayList<>();
this.switchAppendUpsertEnable = switchAppendUpsertEnable;
}
@@ -118,14 +120,16 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
}
@Override
- public void open(Configuration parameters) throws Exception {
+ public void open(Configuration parameters) {
this.subTaskId = getRuntimeContext().getIndexOfThisSubtask();
this.attemptId = getRuntimeContext().getAttemptNumber();
// Initialize the task writer factory.
this.taskWriterFactory.initialize(subTaskId, attemptId);
// Initialize the task writer.
- this.writer = taskWriterFactory.create();
+ createTaskWriter();
+
+ switchHelper = new IcebergModeSwitchHelper(tableSchemaRowType,
incrementalFieldIndex);
// Initialize metric
if (!multipleSink) {
@@ -152,63 +156,56 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
}
}
+ /**
+ * this method should only be called in open() method
+ */
+ private void createTaskWriter() {
+ if (switchAppendUpsertEnable) {
+ // when the job starts and the switch is enabled, the writer
+ // should be in append mode by default
+ taskWriterFactory.switchToAppend();
+ }
+ this.writer = taskWriterFactory.create();
+ }
+
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
// submit the cached write results
- cachedWriteResults.forEach(writeResult -> emit(writeResult));
-
+ LOGGER.info("Submit {} cached write results before checkpoint {}.",
+ cachedWriteResults.size(), checkpointId);
+ cachedWriteResults.forEach(this::emit);
+ cachedWriteResults.clear();
// 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
+ LOGGER.info("close all open file and cache writeResult");
cachedWriteResults.add(writer.complete());
this.writer = taskWriterFactory.create();
}
public void switchToUpsert() throws Exception {
- if (taskWriterFactory.isAppendMode()) {
+ if (!taskWriterFactory.isUpsert()) {
+ LOGGER.info("iceberg writer switch to upsert write mode");
taskWriterFactory.switchToUpsert();
cacheWriteResultAndRecreateWriter();
}
}
- public void switchToAppend() throws Exception {
- if (taskWriterFactory.isAppendMode())
- return;
- taskWriterFactory.switchToAppend();
- cacheWriteResultAndRecreateWriter();
- }
-
@Override
public void processElement(T value) throws Exception {
+
try {
- if (!switchAppendUpsertEnable || multipleSink || metaFieldIndex ==
-1) {
+ if (disableSwitch()) {
writer.write((RowData) value);
- return;
- }
-
- RowData rowData = (RowData) value;
- if (rowData.getBoolean(metaFieldIndex)) {
- switchToUpsert();
} else {
- switchToAppend();
+ if (isIncrementalPhase((RowData) value)) {
+ switchToUpsert();
+ }
+ writer.write(switchHelper.removeIncrementalField((RowData)
value));
}
- writer.write(removeField(rowData, metaFieldIndex,
tableSchemaRowType));
} catch (Exception e) {
if (multipleSink) {
throw e;
@@ -220,6 +217,9 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
}
if (dirtySink != null) {
DirtyData.Builder<Object> builder = DirtyData.builder();
+ if (!disableSwitch()) {
+ value = (T) switchHelper.removeIncrementalField((RowData)
value);
+ }
try {
builder.setData(value)
.setLabels(dirtyOptions.getLabels())
@@ -241,10 +241,27 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
return;
}
if (metricData != null) {
- metricData.invokeWithEstimate(value == null ? "" : value);
+ metricData.invokeWithEstimate(value);
}
}
+ /**
+ * disable switch when the switch property is disabled
+ * or the sink is multiple sink (the data is in json format)
+ * or the incremental field is not set
+ */
+ private boolean disableSwitch() {
+ return !switchAppendUpsertEnable || multipleSink ||
incrementalFieldIndex == DEFAULT_META_INDEX;
+ }
+
+ /**
+ * check if the data is incremental phase
+ * by checking the incremental field
+ */
+ private boolean isIncrementalPhase(RowData rowData) {
+ return rowData.getBoolean(incrementalFieldIndex);
+ }
+
@Override
public void initializeState(FunctionInitializationContext context) throws
Exception {
// init metric state
@@ -310,6 +327,9 @@ public class IcebergSingleStreamWriter<T> extends
IcebergProcessFunction<T, Writ
}
private void emit(WriteResult result) {
+ LOGGER.debug("Emit iceberg write result dataFiles: {},
result.deleteFiles {}",
+ result.dataFiles(), result.deleteFiles());
collector.collect(result);
}
+
}
