apache / apache/iceberg

Merge Into Statement Lossing data on Non-partitioned table

Open
#17,037 3 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

### Apache Iceberg version

1.4.0

### Query engine

Spark

### Please describe the bug 🐞

We've been using iceberg since long now on AWS, but came across the following issue recently in our Production environment.

With Upsert query using MERGE INTO stetement, we observed for one of the iceberg table (non-partitioned), while rewriting the data files, it has removed some of the records. We have checked with the time-travel and the records were present earlier, but in one the of rewrite history, the records got removed from the table.

The table here is in silver layer, where we're merging data from different data sources into single table, to identify the records, we've "datasource_id" column present along with some system columns like created_on_dt, updated_on_dt, w_inserted_dt, w_updated_dt, etl_run_id etc.

We've never observed any such issue with Partitioned tables though. Has anyone came across similar issue, I want to understand why iceberg is removing/deleting the unchanged records?

Following is the merge query we use :
```
upsert_sql = f"""
MERGE INTO {table} AS target
USING input_data AS source
ON concat(coalesce(target.integration_id, 'UNKNOWN'), coalesce(target.datasource_id, 'DFLT')) = concat(coalesce(source.integration_id, 'UNKNOWN'), coalesce(source.datasource_id, 'DFLT'))
WHEN MATCHED THEN
UPDATE SET {', '.join(f"target.{col} = source.{col}" for col in input_df.columns if col not in exclude_columns)}
WHEN NOT MATCHED THEN
INSERT ({', '.join(input_df.columns)})
VALUES ({', '.join(f"source.{col}" for col in input_df.columns)})
"""
# Execute the upsert SQL statement
spark.sql(upsert_sql)

cleaned_df = cleaned_df.dropDuplicates(["integration_id"])
upsert_data_iceberg(spark, cleaned_df, sl_athena_output_db_name, iceberg_table_name, s3_silver_output_location, logger)
```

Before snapshot:

snapshot-id = 5244992060615715741
total-records = 99,375,307
total-data-files = 936

Then the next snapshot:

snapshot-id = 7953544942377759370

added-records = 49,077,409
deleted-records = 55,719,302

added-data-files = 4
deleted-data-files = 38

total-records = 92,733,414

Net loss:

99,375,307
-92,733,414
------------
6,641,893 rows

### Willingness to contribute

- [ ] I can contribute a fix for this bug independently
- [x] 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

Contributor guide

Open the contributing guide

Research direction

Start with the supplied Spark MERGE INTO upsert against the non-partitioned Iceberg table, then compare the reported snapshots and time-travel results. Narrow down which rewrite produces the 6,641,893-row loss and establish a minimal reproduction. Done means the cause is confirmed and unchanged records no longer disappear.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark
Domain
data-engineering, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.