DynamicIcebergSink writes not being read by IcebergSource streaming

[bezdomniy]

Apache Iceberg version

1.10.0 (latest release)

Query engine

Flink

Please describe the bug 🐞

Hi all,
Have 2 Flink (v1.20.3) jobs - one is writing records to an Iceberg table with org.apache.iceberg.flink.sink.FlinkSink, and another streaming incremental changes from the table with org.apache.iceberg.flink.source.IcebergSource. This was working fine.

I then changed FlinkSink to org.apache.iceberg.flink.sink.dynamic.DynamicIcebergSink. The writes to the Iceberg table are still working fine - however now the incremental reads do not get picked up by the IcebergSource in the second Flink job. The IcebergSource recognises the new snapshots, but says: "Discovered 0 splits from incremental scan". When I query the difference between the snapshots with Athena however, there definitely are incremental records there.

I couldn't find any documentation on things to consider when moving to DynamicIcebergSink - is there something that I'm missing here?

My sink is defined like this:

DynamicIcebergSink.Builder<GenericRecord> flinkSinkBuilder =
        DynamicIcebergSink.forInput(dataStream).generator((inputRecord, out) -> {
            var record = new DynamicRecord(TableIdentifier.of(s3_table_db, s3_table_name),
                    "main", icebergSchema,
                    avroGenericRecordToRowDataMapper.map(inputRecord), partitionSpec,
                    partitionSpec.isPartitioned() ? DistributionMode.HASH
                            : DistributionMode.NONE,
                    2);
            record.setUpsertMode(false);
            out.collect(record);
        }).catalogLoader(icebergCatalogLoader).immediateTableUpdate(true)
                .set("write.upsert.enabled", "false");

Thank you.

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[pvary]

The Flink connector source is only reading append only commits. If there are updates in a commit, then it is ignored by the source.

My guess is that you used the old sink in append mode.

I see that you have set record.setUpsertMode(false). Could you please check the type of the files added to the table by the new DynamicSink?

[bezdomniy]

The type of the files? Do you mean format? For example -
SELECT * FROM "default"."prices_s3table$manifests" - gives me a list of avro files for each snapshot id
SELECT * FROM "default"."prices_s3table$files" - files me a list of parquet files

[pvary]

I would like to confirm that you have only append commits in the target table. If you have delete files in the commit then it is not an append commit, and the reader will skip them

[bezdomniy]

When I query the files in the table, and content column is all 0 value - based on the docs that means: "0 - Data"

[pvary]

If the number of files in the table keeps increasing then it seems like an issue with the Source configuration.

[bezdomniy]

It would seem to be that way - but the same source config works when its FlinkSink writing to the table, but not when it is DynamicIcebergSink. I'm trying to find what is different.

The other thing is that StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTAL will read all existing rows, but not any incremental rows.

My source is like this:

var source = IcebergSource.forOutputType(avroRecordConverter).tableLoader(tableLoader)
        .assignerFactory(new SimpleSplitAssignerFactory()).streaming(true)
        .streamingStartingStrategy(
                StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
        .monitorInterval(Duration.ofSeconds(10)).branch("main");

[pvary]

Could you check the snapshots: https://iceberg.apache.org/docs/latest/spark-queries/#snapshots?

[pvary]

My next idea is that we create RowDelta commits, even if we don't have deletes.

In the original sink we checked and created append commits whenever it was possible:

iceberg/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergCommitter.java

Lines 241 to 253 in c22bac8

	if (summary.deleteFilesCount() == 0) {
	// To be compatible with iceberg format V1.
	AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
	for (WriteResult result : pendingResults.values()) {
	Preconditions.checkState(
	result.referencedDataFiles().length == 0,
	"Should have no referenced data files for append.");
	Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
	}
	String description = "append";
	logCommitSummary(summary, description);
	// fail all commits as really its only one
	commitOperation(appendFiles, description, newFlinkJobId, operatorId, checkpointId);

In the Dynamic sink:

iceberg/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

Lines 309 to 329 in c22bac8

	RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
	for (WriteResult result : writeResults) {
	// Row delta validations are not needed for streaming changes that write equality deletes.
	// Equality deletes are applied to data in all previous sequence numbers, so retries may
	// push deletes further in the future, but do not affect correctness. Position deletes
	// committed to the table in this path are used only to delete rows from data files that are
	// being added in this commit. There is no way for data files added along with the delete
	// files to be concurrently removed, so there is no need to validate the files referenced by
	// the position delete files that are being committed.
	Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
	Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
	}
	// Every Flink checkpoint contains a set of independent changes which can be committed
	// together. While it is technically feasible to combine append-only data across checkpoints,
	// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
	// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
	// intervals or when concurrent checkpointing is enabled.
	commitOperation(
	table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
	}

[bezdomniy]

Ah yes - the first snapshot operation is append, and subsequent ones are overwrite. Is there any way to disable overwrite in DynamicIcebergSink?

[pvary]

Sadly, we need a patch for this.
A similar if that we have in the original sink

[pvary]

If you have time to create a PR, I would be happy to review!

[bezdomniy]

I will give it a try - thank you for your help.