Repository: apache/iceberg-go
Pull request #356 (2025-03-21 18:10:26)

feat(table): add ReplaceFiles method to transaction
Implementing a simple `ReplaceDataFiles` method for transactions to create an `Overwrite` operation for deleting/adding files in a single operation for use with compaction operations or otherwise replacing one or more data files with new data files.

zeroshade: @arnaudbriche Can you take a look at this and tell me what you think? Along with helping me to create some test cases?
arnaudbriche: Hi @zeroshade 

I just took a look at this PR and it seems good.
The only missing things to me is ensuring all `filesToDelete` are actually part of the previous snapshot.

As for the tests, let me know how I can contribute and I'll do.
Here are the steps I used to validate my draft of an implementation, and the result with the proposed one:

1. Execute the following Clickhouse query 5 times 

```sql
insert into function s3('http://localhost:9001/test01/table_test_01/data/data.parquet', 'minio', 'minio123')
select
    *
from (
    select
        *
    from generateRandom('
        date Date,
        name String,
        value Float64
    ')
    limit 10000000
)
settings s3_create_new_file_on_insert=true
```
It generates four parquet files named data.1.parquet, data.2.parquet, .... filled with 10 million random rows.

An S3 list gives:

```
minio://test01/table_test_01/data/data.1.parquet 190 MB 2025-03-20 14:40:46.243 +0000 UTC "8980e1317bd85ef9dd119c9204794365-11"
minio://test01/table_test_01/data/data.2.parquet 190 MB 2025-03-20 14:40:49.381 +0000 UTC "ae7965d72849b525440400adaa443110-11"
minio://test01/table_test_01/data/data.3.parquet 190 MB 2025-03-20 17:27:08.531 +0000 UTC "c89f9c2e99841cfe9215d32cc2fdf368-11"
minio://test01/table_test_01/data/data.4.parquet 190 MB 2025-03-20 17:27:16.633 +0000 UTC "95ed4f51b635fb59b94c2f4bb1765eef-11"
minio://test01/table_test_01/data/data.parquet 190 MB 2025-03-20 14:40:28.308 +0000 UTC "a8d4edc6dd647d156bd02e21c147353d-11"
```


2. Run some code using `AddFiles` from this package. Once for each Parquet file

Now the S3 listing gives

```
minio://test01/table_test_01/data/data.1.parquet 190 MB 2025-03-20 14:40:46.243 +0000 UTC "8980e1317bd85ef9dd119c9204794365-11"
minio://test01/table_test_01/data/data.2.parquet 190 MB 2025-03-20 14:40:49.381 +0000 UTC "ae7965d72849b525440400adaa443110-11"
minio://test01/table_test_01/data/data.3.parquet 190 MB 2025-03-20 17:27:08.531 +0000 UTC "c89f9c2e99841cfe9215d32cc2fdf368-11"
minio://test01/table_test_01/data/data.4.parquet 190 MB 2025-03-20 17:27:16.633 +0000 UTC "95ed4f51b635fb59b94c2f4bb1765eef-11"
minio://test01/table_test_01/data/data.parquet 190 MB 2025-03-20 14:40:28.308 +0000 UTC "a8d4edc6dd647d156bd02e21c147353d-11"
minio://test01/table_test_01/metadata/000000000000-0195c3f3-468b-7e11-afb8-bbdfddf1a59e.metadata.json 613 B 2025-03-23 17:00:55.309 +0000 UTC "d219ebb2a78eab9f0469e0ee0db9038b"
minio://test01/table_test_01/metadata/000000000001-0195c3f3-46a0-7ce0-a3fe-59196edaa285.metadata.json 1.5 kB 2025-03-23 17:00:55.329 +0000 UTC "e874ee3387d99ad171d363b23988329b"
minio://test01/table_test_01/metadata/000000000002-0195c3f3-66b7-71f0-8948-84bf29865c57.metadata.json 2.1 kB 2025-03-23 17:01:03.543 +0000 UTC "2fd8d023298234ffb53b5fbc5c038a3b"
minio://test01/table_test_01/metadata/000000000003-0195c3f3-7dc5-7674-b93d-575b4d5da3c6.metadata.json 2.7 kB 2025-03-23 17:01:09.445 +0000 UTC "63bde62ccfa23826c5efd8059f133151"
minio://test01/table_test_01/metadata/000000000004-0195c3f3-ab3c-73a1-bc34-63562c02ddff.metadata.json 3.3 kB 2025-03-23 17:01:21.085 +0000 UTC "ab58765d0dfc77dbeaac737bcb254aed"
minio://test01/table_test_01/metadata/000000000005-0195c3f3-bf61-7a8b-983c-c92ad28bbb02.metadata.json 3.9 kB 2025-03-23 17:01:26.244 +0000 UTC "e2419b2bc0ce3647121b9587caae92af"
minio://test01/table_test_01/metadata/4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339-m1.avro 3.7 kB 2025-03-23 17:00:55.324 +0000 UTC "7e4cf98cdc8d211720093a7842f851c5"
minio://test01/table_test_01/metadata/ea98f7f9-5986-48c3-a9b7-54750d4e6156-m1.avro 3.7 kB 2025-03-23 17:01:09.436 +0000 UTC "07a0cf047550a3c26f24db78b0db2cb3"
minio://test01/table_test_01/metadata/f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01-m1.avro 3.7 kB 2025-03-23 17:01:21.068 +0000 UTC "7b8e1878dd7dc656528d46c45ef0d451"
minio://test01/table_test_01/metadata/f66ff9e0-f661-42b7-a437-c9eafb7ae559-m1.avro 3.7 kB 2025-03-23 17:01:26.227 +0000 UTC "71e2b7f5b704f947942cbc77059c74d8"
minio://test01/table_test_01/metadata/fdbed012-461c-47e1-966f-74e68138c5ce-m1.avro 3.7 kB 2025-03-23 17:01:03.534 +0000 UTC "35155acb992477a12667e3d8680930f4"
minio://test01/table_test_01/metadata/snap-3693479042177426923-0-fdbed012-461c-47e1-966f-74e68138c5ce.avro 2.2 kB 2025-03-23 17:01:03.54 +0000 UTC "85c7fb383a1f58950378df693b273bf1"
minio://test01/table_test_01/metadata/snap-5504984771894290663-0-ea98f7f9-5986-48c3-a9b7-54750d4e6156.avro 2.3 kB 2025-03-23 17:01:09.442 +0000 UTC "8174888235180e5775a9e6d1b8121932"
minio://test01/table_test_01/metadata/snap-5947020003963383724-0-f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01.avro 2.3 kB 2025-03-23 17:01:21.073 +0000 UTC "ff128ba897fe259fca3c551e6269c35e"
minio://test01/table_test_01/metadata/snap-7180314873379640306-0-f66ff9e0-f661-42b7-a437-c9eafb7ae559.avro 2.4 kB 2025-03-23 17:01:26.236 +0000 UTC "37dec188713bf208d1e8b1ab3d709909"
minio://test01/table_test_01/metadata/snap-9068225639057837698-0-4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339.avro 2.2 kB 2025-03-23 17:00:55.327 +0000 UTC "2a0841ee7d1a3fba6aa0f80ff1e6d4e3"
minio://test01/table_test_01/metadata/version-hint.text 63 B 2025-03-23 17:01:26.247 +0000 UTC "0b120f62db0059f2001c3559f9e4ae10"
```

And the following SQL query in clickhouse-local gives the right result

```sql
:) select count(*), groupUniqArray(_file) from iceberg('http://localhost:9001/test01/table_test_01', 'minio', 'minio123')

SELECT
    count(*),
    groupUniqArray(_file)
FROM iceberg('http://localhost:9001/test01/table_test_01', 'minio', 'minio123')

Query id: 9d198fcc-58a3-4b46-bf3c-724839a6c11c

   ┌──count()─┬─groupUniqArray(_file)────────────────────────────────────────────────────────────────┐
1. │ 50000000 │ ['data.4.parquet','data.1.parquet','data.3.parquet','data.2.parquet','data.parquet'] │
   └──────────┴──────────────────────────────────────────────────────────────────────────────────────┘

1 row in set. Elapsed: 0.147 sec. Processed 50.00 million rows, 0.00 B (339.92 million rows/s., 0.00 B/s.)
Peak memory usage: 265.58 MiB.
````

3. Merge 2 of the table files with clickhouse


```sql
insert into function s3('http://localhost:9001/test01/table_test_01/data/data.parquet', 'minio', 'minio123')
select
    *
from s3('http://localhost:9001/test01/table_test_01/data/data.{1..2}.parquet', 'minio', 'minio123')
settings
    s3_create_new_file_on_insert=true,
    schema_inference_make_columns_nullable='auto'
```

S3 listing looks right:

```
minio://test01/table_test_01/data/data.1.parquet 190 MB 2025-03-20 14:40:46.243 +0000 UTC "8980e1317bd85ef9dd119c9204794365-11"
minio://test01/table_test_01/data/data.2.parquet 190 MB 2025-03-20 14:40:49.381 +0000 UTC "ae7965d72849b525440400adaa443110-11"
minio://test01/table_test_01/data/data.3.parquet 190 MB 2025-03-20 17:27:08.531 +0000 UTC "c89f9c2e99841cfe9215d32cc2fdf368-11"
minio://test01/table_test_01/data/data.4.parquet 190 MB 2025-03-20 17:27:16.633 +0000 UTC "95ed4f51b635fb59b94c2f4bb1765eef-11"
minio://test01/table_test_01/data/data.5.parquet 381 MB 2025-03-23 17:04:42.863 +0000 UTC "1512b41da74c4f59dd0e97bceb049046-22"
minio://test01/table_test_01/data/data.parquet 190 MB 2025-03-20 14:40:28.308 +0000 UTC "a8d4edc6dd647d156bd02e21c147353d-11"
minio://test01/table_test_01/metadata/000000000000-0195c3f3-468b-7e11-afb8-bbdfddf1a59e.metadata.json 613 B 2025-03-23 17:00:55.309 +0000 UTC "d219ebb2a78eab9f0469e0ee0db9038b"
minio://test01/table_test_01/metadata/000000000001-0195c3f3-46a0-7ce0-a3fe-59196edaa285.metadata.json 1.5 kB 2025-03-23 17:00:55.329 +0000 UTC "e874ee3387d99ad171d363b23988329b"
minio://test01/table_test_01/metadata/000000000002-0195c3f3-66b7-71f0-8948-84bf29865c57.metadata.json 2.1 kB 2025-03-23 17:01:03.543 +0000 UTC "2fd8d023298234ffb53b5fbc5c038a3b"
minio://test01/table_test_01/metadata/000000000003-0195c3f3-7dc5-7674-b93d-575b4d5da3c6.metadata.json 2.7 kB 2025-03-23 17:01:09.445 +0000 UTC "63bde62ccfa23826c5efd8059f133151"
minio://test01/table_test_01/metadata/000000000004-0195c3f3-ab3c-73a1-bc34-63562c02ddff.metadata.json 3.3 kB 2025-03-23 17:01:21.085 +0000 UTC "ab58765d0dfc77dbeaac737bcb254aed"
minio://test01/table_test_01/metadata/000000000005-0195c3f3-bf61-7a8b-983c-c92ad28bbb02.metadata.json 3.9 kB 2025-03-23 17:01:26.244 +0000 UTC "e2419b2bc0ce3647121b9587caae92af"
minio://test01/table_test_01/metadata/4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339-m1.avro 3.7 kB 2025-03-23 17:00:55.324 +0000 UTC "7e4cf98cdc8d211720093a7842f851c5"
minio://test01/table_test_01/metadata/ea98f7f9-5986-48c3-a9b7-54750d4e6156-m1.avro 3.7 kB 2025-03-23 17:01:09.436 +0000 UTC "07a0cf047550a3c26f24db78b0db2cb3"
minio://test01/table_test_01/metadata/f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01-m1.avro 3.7 kB 2025-03-23 17:01:21.068 +0000 UTC "7b8e1878dd7dc656528d46c45ef0d451"
minio://test01/table_test_01/metadata/f66ff9e0-f661-42b7-a437-c9eafb7ae559-m1.avro 3.7 kB 2025-03-23 17:01:26.227 +0000 UTC "71e2b7f5b704f947942cbc77059c74d8"
minio://test01/table_test_01/metadata/fdbed012-461c-47e1-966f-74e68138c5ce-m1.avro 3.7 kB 2025-03-23 17:01:03.534 +0000 UTC "35155acb992477a12667e3d8680930f4"
minio://test01/table_test_01/metadata/snap-3693479042177426923-0-fdbed012-461c-47e1-966f-74e68138c5ce.avro 2.2 kB 2025-03-23 17:01:03.54 +0000 UTC "85c7fb383a1f58950378df693b273bf1"
minio://test01/table_test_01/metadata/snap-5504984771894290663-0-ea98f7f9-5986-48c3-a9b7-54750d4e6156.avro 2.3 kB 2025-03-23 17:01:09.442 +0000 UTC "8174888235180e5775a9e6d1b8121932"
minio://test01/table_test_01/metadata/snap-5947020003963383724-0-f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01.avro 2.3 kB 2025-03-23 17:01:21.073 +0000 UTC "ff128ba897fe259fca3c551e6269c35e"
minio://test01/table_test_01/metadata/snap-7180314873379640306-0-f66ff9e0-f661-42b7-a437-c9eafb7ae559.avro 2.4 kB 2025-03-23 17:01:26.236 +0000 UTC "37dec188713bf208d1e8b1ab3d709909"
minio://test01/table_test_01/metadata/snap-9068225639057837698-0-4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339.avro 2.2 kB 2025-03-23 17:00:55.327 +0000 UTC "2a0841ee7d1a3fba6aa0f80ff1e6d4e3"
minio://test01/table_test_01/metadata/version-hint.text 63 B 2025-03-23 17:01:26.247 +0000 UTC "0b120f62db0059f2001c3559f9e4ae10"
```

4. Use the new `Replacefiles` method to update metadata:

S3 looks OK:

```
minio://test01/table_test_01/data/data.1.parquet 190 MB 2025-03-20 14:40:46.243 +0000 UTC "8980e1317bd85ef9dd119c9204794365-11"
minio://test01/table_test_01/data/data.2.parquet 190 MB 2025-03-20 14:40:49.381 +0000 UTC "ae7965d72849b525440400adaa443110-11"
minio://test01/table_test_01/data/data.3.parquet 190 MB 2025-03-20 17:27:08.531 +0000 UTC "c89f9c2e99841cfe9215d32cc2fdf368-11"
minio://test01/table_test_01/data/data.4.parquet 190 MB 2025-03-20 17:27:16.633 +0000 UTC "95ed4f51b635fb59b94c2f4bb1765eef-11"
minio://test01/table_test_01/data/data.5.parquet 381 MB 2025-03-23 17:04:42.863 +0000 UTC "1512b41da74c4f59dd0e97bceb049046-22"
minio://test01/table_test_01/data/data.parquet 190 MB 2025-03-20 14:40:28.308 +0000 UTC "a8d4edc6dd647d156bd02e21c147353d-11"
minio://test01/table_test_01/metadata/000000000000-0195c3f3-468b-7e11-afb8-bbdfddf1a59e.metadata.json 613 B 2025-03-23 17:00:55.309 +0000 UTC "d219ebb2a78eab9f0469e0ee0db9038b"
minio://test01/table_test_01/metadata/000000000001-0195c3f3-46a0-7ce0-a3fe-59196edaa285.metadata.json 1.5 kB 2025-03-23 17:00:55.329 +0000 UTC "e874ee3387d99ad171d363b23988329b"
minio://test01/table_test_01/metadata/000000000002-0195c3f3-66b7-71f0-8948-84bf29865c57.metadata.json 2.1 kB 2025-03-23 17:01:03.543 +0000 UTC "2fd8d023298234ffb53b5fbc5c038a3b"
minio://test01/table_test_01/metadata/000000000003-0195c3f3-7dc5-7674-b93d-575b4d5da3c6.metadata.json 2.7 kB 2025-03-23 17:01:09.445 +0000 UTC "63bde62ccfa23826c5efd8059f133151"
minio://test01/table_test_01/metadata/000000000004-0195c3f3-ab3c-73a1-bc34-63562c02ddff.metadata.json 3.3 kB 2025-03-23 17:01:21.085 +0000 UTC "ab58765d0dfc77dbeaac737bcb254aed"
minio://test01/table_test_01/metadata/000000000005-0195c3f3-bf61-7a8b-983c-c92ad28bbb02.metadata.json 3.9 kB 2025-03-23 17:01:26.244 +0000 UTC "e2419b2bc0ce3647121b9587caae92af"
minio://test01/table_test_01/metadata/000000000006-0195c3f7-eb41-72da-8b82-6112e65326c5.metadata.json 4.6 kB 2025-03-23 17:05:59.618 +0000 UTC "438740422277ec4341928cff34e0d17c"
minio://test01/table_test_01/metadata/4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339-m1.avro 3.7 kB 2025-03-23 17:00:55.324 +0000 UTC "7e4cf98cdc8d211720093a7842f851c5"
minio://test01/table_test_01/metadata/d50323c7-6dad-4850-8a73-1416701e8d50-m1.avro 3.7 kB 2025-03-23 17:05:59.571 +0000 UTC "ab901aaee0534ac396485c3de2dadcbc"
minio://test01/table_test_01/metadata/d50323c7-6dad-4850-8a73-1416701e8d50-m2.avro 3.8 kB 2025-03-23 17:05:59.588 +0000 UTC "2a831fdd897dfbcc4694cd64d4a951f0"
minio://test01/table_test_01/metadata/ea98f7f9-5986-48c3-a9b7-54750d4e6156-m1.avro 3.7 kB 2025-03-23 17:01:09.436 +0000 UTC "07a0cf047550a3c26f24db78b0db2cb3"
minio://test01/table_test_01/metadata/f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01-m1.avro 3.7 kB 2025-03-23 17:01:21.068 +0000 UTC "7b8e1878dd7dc656528d46c45ef0d451"
minio://test01/table_test_01/metadata/f66ff9e0-f661-42b7-a437-c9eafb7ae559-m1.avro 3.7 kB 2025-03-23 17:01:26.227 +0000 UTC "71e2b7f5b704f947942cbc77059c74d8"
minio://test01/table_test_01/metadata/fdbed012-461c-47e1-966f-74e68138c5ce-m1.avro 3.7 kB 2025-03-23 17:01:03.534 +0000 UTC "35155acb992477a12667e3d8680930f4"
minio://test01/table_test_01/metadata/snap-3693479042177426923-0-fdbed012-461c-47e1-966f-74e68138c5ce.avro 2.2 kB 2025-03-23 17:01:03.54 +0000 UTC "85c7fb383a1f58950378df693b273bf1"
minio://test01/table_test_01/metadata/snap-4223232085747188730-0-d50323c7-6dad-4850-8a73-1416701e8d50.avro 2.3 kB 2025-03-23 17:05:59.614 +0000 UTC "a34f57bcd01ca82ea277316b673bb90b"
minio://test01/table_test_01/metadata/snap-5504984771894290663-0-ea98f7f9-5986-48c3-a9b7-54750d4e6156.avro 2.3 kB 2025-03-23 17:01:09.442 +0000 UTC "8174888235180e5775a9e6d1b8121932"
minio://test01/table_test_01/metadata/snap-5947020003963383724-0-f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01.avro 2.3 kB 2025-03-23 17:01:21.073 +0000 UTC "ff128ba897fe259fca3c551e6269c35e"
minio://test01/table_test_01/metadata/snap-7180314873379640306-0-f66ff9e0-f661-42b7-a437-c9eafb7ae559.avro 2.4 kB 2025-03-23 17:01:26.236 +0000 UTC "37dec188713bf208d1e8b1ab3d709909"
minio://test01/table_test_01/metadata/snap-9068225639057837698-0-4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339.avro 2.2 kB 2025-03-23 17:00:55.327 +0000 UTC "2a0841ee7d1a3fba6aa0f80ff1e6d4e3"
minio://test01/table_test_01/metadata/version-hint.text 63 B 2025-03-23 17:05:59.62 +0000 UTC "e47a428b023a9c5c9c578fe4ff9e59ab"
```

And Clickhouse test query results look good:

```sql
:) select count(*), groupUniqArray(_file) from iceberg('http://localhost:9001/test01/table_test_01', 'minio', 'minio123')

SELECT
    count(*),
    groupUniqArray(_file)
FROM iceberg('http://localhost:9001/test01/table_test_01', 'minio', 'minio123')

Query id: 3e476f8d-69c6-4610-b9e3-671e0e015b98

   ┌──count()─┬─groupUniqArray(_file)───────────────────────────────────────────────┐
1. │ 50000000 │ ['data.4.parquet','data.5.parquet','data.3.parquet','data.parquet'] │
   └──────────┴─────────────────────────────────────────────────────────────────────┘

1 row in set. Elapsed: 0.222 sec. Processed 50.00 million rows, 0.00 B (225.21 million rows/s., 0.00 B/s.)
Peak memory usage: 787.14 MiB.
```

Everything looked great.
Until, I took a look at the last metadata.json file, whose content is below:

```json 
{
    "last-sequence-number": 6,
    "format-version": 2,
    "table-uuid": "0195c3f3-468b-7d27-a957-53e1ce86699d",
    "location": "s3://test01/table_test_01",
    "last-updated-ms": 1742749559614,
    "last-column-id": 3,
    "schemas": [
        {
            "type": "struct",
            "fields": [
                {
                    "type": "date",
                    "id": 1,
                    "name": "date",
                    "required": true
                },
                {
                    "type": "string",
                    "id": 2,
                    "name": "name",
                    "required": true
                },
                {
                    "type": "double",
                    "id": 3,
                    "name": "value",
                    "required": true
                }
            ],
            "schema-id": 0,
            "identifier-field-ids": []
        }
    ],
    "current-schema-id": 0,
    "partition-specs": [
        {
            "spec-id": 0,
            "fields": []
        }
    ],
    "default-spec-id": 0,
    "last-partition-id": 999,
    "properties": {
        "schema.name-mapping.default": "[{\"names\":[\"date\"],\"field-id\":1},{\"names\":[\"name\"],\"field-id\":2},{\"names\":[\"value\"],\"field-id\":3}]"
    },
    "snapshots": [
        {
            "snapshot-id": 9068225639057837698,
            "sequence-number": 1,
            "timestamp-ms": 1742749255326,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-9068225639057837698-0-4ee7e614-6eb9-4ffe-8410-c5a0fbfe3339.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "190503335",
                "added-records": "10000000",
                "operation": "append",
                "total-data-files": "1",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "190503335",
                "total-position-deletes": "0",
                "total-records": "10000000"
            },
            "schema-id": 0
        },
        {
            "snapshot-id": 3693479042177426923,
            "parent-snapshot-id": 9068225639057837698,
            "sequence-number": 2,
            "timestamp-ms": 1742749263540,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-3693479042177426923-0-fdbed012-461c-47e1-966f-74e68138c5ce.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "190499007",
                "added-records": "10000000",
                "operation": "append",
                "total-data-files": "2",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "381002342",
                "total-position-deletes": "0",
                "total-records": "20000000"
            },
            "schema-id": 0
        },
        {
            "snapshot-id": 5504984771894290663,
            "parent-snapshot-id": 3693479042177426923,
            "sequence-number": 3,
            "timestamp-ms": 1742749269442,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-5504984771894290663-0-ea98f7f9-5986-48c3-a9b7-54750d4e6156.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "190500025",
                "added-records": "10000000",
                "operation": "append",
                "total-data-files": "3",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "571502367",
                "total-position-deletes": "0",
                "total-records": "30000000"
            },
            "schema-id": 0
        },
        {
            "snapshot-id": 5947020003963383724,
            "parent-snapshot-id": 5504984771894290663,
            "sequence-number": 4,
            "timestamp-ms": 1742749281073,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-5947020003963383724-0-f2ccf44f-5c74-42ba-8a3f-aa79b6c17e01.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "190511181",
                "added-records": "10000000",
                "operation": "append",
                "total-data-files": "4",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "762013548",
                "total-position-deletes": "0",
                "total-records": "40000000"
            },
            "schema-id": 0
        },
        {
            "snapshot-id": 7180314873379640306,
            "parent-snapshot-id": 5947020003963383724,
            "sequence-number": 5,
            "timestamp-ms": 1742749286236,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-7180314873379640306-0-f66ff9e0-f661-42b7-a437-c9eafb7ae559.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "190495936",
                "added-records": "10000000",
                "operation": "append",
                "total-data-files": "5",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "952509484",
                "total-position-deletes": "0",
                "total-records": "50000000"
            },
            "schema-id": 0
        },
        {
            "snapshot-id": 4223232085747188730,
            "parent-snapshot-id": 7180314873379640306,
            "sequence-number": 6,
            "timestamp-ms": 1742749559614,
            "manifest-list": "s3://test01/table_test_01/metadata/snap-4223232085747188730-0-d50323c7-6dad-4850-8a73-1416701e8d50.avro",
            "summary": {
                "added-data-files": "1",
                "added-files-size": "381001164",
                "added-records": "20000000",
                "deleted-data-files": "5",
                "deleted-records": "50000000",
                "operation": "overwrite",
                "removed-files-size": "952509484",
                "total-data-files": "1",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "381001164",
                "total-position-deletes": "0",
                "total-records": "20000000"
            },
            "schema-id": 0
        }
    ],
    "current-snapshot-id": 4223232085747188730,
    "snapshot-log": [
        {
            "snapshot-id": 9068225639057837698,
            "timestamp-ms": 1742749255326
        },
        {
            "snapshot-id": 3693479042177426923,
            "timestamp-ms": 1742749263540
        },
        {
            "snapshot-id": 5504984771894290663,
            "timestamp-ms": 1742749269442
        },
        {
            "snapshot-id": 5947020003963383724,
            "timestamp-ms": 1742749281073
        },
        {
            "snapshot-id": 7180314873379640306,
            "timestamp-ms": 1742749286236
        },
        {
            "snapshot-id": 4223232085747188730,
            "timestamp-ms": 1742749559614
        }
    ],
    "sort-orders": [
        {
            "order-id": 0,
            "fields": []
        }
    ],
    "default-sort-order-id": 0,
    "refs": {
        "main": {
            "snapshot-id": 4223232085747188730,
            "type": "branch"
        }
    }
}
```

The summary for the last snapshot is not ok.

```json
        "summary": {
                "added-data-files": "1",
                "added-files-size": "381001164",
                "added-records": "20000000",
                "deleted-data-files": "5",
                "deleted-records": "50000000",
                "operation": "overwrite",
                "removed-files-size": "952509484",
                "total-data-files": "1",
                "total-delete-files": "0",
                "total-equality-deletes": "0",
                "total-files-size": "381001164",
                "total-position-deletes": "0",
                "total-records": "20000000"
            }
```

We should have:
- total-records = 50000000
- deleted-data-files = 2
- deleted-records = 20000000
- total-data-files = 4



zeroshade: Hey @arnaudbriche I've found the issue and fixed it for the snapshot summary to be correct now. I'll probably work more on creating some tests tomorrow. Likely based on your example :smile:
arnaudbriche: Hi @zeroshade ,

I confirm your latest commit fixed the issue in my tests.
I think it's also important to tests scenarios where some of the removed datafiles are part of a manifest containing non-removed files.

Test is nearly same as above, but files are not appended 1 by 1.

I tested with this sequence:
1. append [data.parquet, data.1.parquet]
2. append [data.2.parquet, data.3.parquet]
3. append [data.4. parquet]
4. replace [data.4.parquet=data.1.parquet+data.2.parquet]

And everything works as expected !
arnaudbriche: > The only missing things to me is ensuring all `filesToDelete` are actually part of the previous snapshot.

Regading this, I created a small PR on your fork that just adds the check: https://github.com/zeroshade/iceberg-go/pull/1
arnaudbriche: Hi @zeroshade ,

Not really related to this specific PR. But are it seemed implemented in the commit your pointed me to.
Is really anything in the lib that allow pruning snapshots ? 
For now, the only issue I have in my use-case is that I found no way to prevent unlimited growth of snapshot history.
arnaudbriche: I'll be happy to help with that!
Just to be clear, I'm talking about something like: https://iceberg.apache.org/javadoc/1.8.1/org/apache/iceberg/ExpireSnapshots.html

Patch:
diff --git a/config/config.go b/config/config.go
index d4cf88a..3d645dd 100644
--- a/config/config.go
+++ b/config/config.go
@@ -25,12 +25,14 @@ import (
 )
 
 const (
-	cfgFile = ".iceberg-go.yaml"
+	cfgFile           = ".iceberg-go.yaml"
+	defaultMaxWorkers = 5
 )
 
 type Config struct {
 	DefaultCatalog string                   `yaml:"default-catalog"`
 	Catalogs       map[string]CatalogConfig `yaml:"catalog"`
+	MaxWorkers     int                      `yaml:"max-workers"`
 }
 
 type CatalogConfig struct {
@@ -88,6 +90,9 @@ func fromConfigFiles() Config {
 	if cfg.DefaultCatalog == "" {
 		cfg.DefaultCatalog = "default"
 	}
+	if cfg.MaxWorkers <= 0 {
+		cfg.MaxWorkers = defaultMaxWorkers
+	}
 
 	return cfg
 }
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 5a53b6a..3e67d8b 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -23,7 +23,9 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"iter"
 	"regexp"
+	"runtime"
 	"strconv"
 	"strings"
 	"unicode/utf8"
@@ -32,6 +34,7 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/decimal"
 	"github.com/apache/arrow-go/v18/arrow/decimal128"
 	"github.com/apache/iceberg-go"
+	"golang.org/x/sync/errgroup"
 )
 
 // Enumerated is a quick way to represent a sequenced value that can
@@ -478,3 +481,59 @@ func TruncateUpperBoundBinary(val []byte, trunc int) []byte {
 
 	return nil
 }
+
+func MapExec[T, S any](nWorkers int, slice []T, fn func(T) (S, error)) iter.Seq2[S, error] {
+	if nWorkers <= 0 {
+		nWorkers = runtime.GOMAXPROCS(0)
+	}
+
+	nWorkers = min(nWorkers, len(slice))
+
+	var g errgroup.Group
+	ch := make(chan T, len(slice))
+	out := make(chan S, nWorkers)
+
+	for range nWorkers {
+		g.Go(func() error {
+			for v := range ch {
+				result, err := fn(v)
+				if err != nil {
+					return err
+				}
+				out <- result
+			}
+
+			return nil
+		})
+	}
+
+	for _, v := range slice {
+		ch <- v
+	}
+	close(ch)
+
+	var err error
+	go func() {
+		defer close(out)
+		err = g.Wait()
+	}()
+
+	return func(yield func(S, error) bool) {
+		defer func() {
+			// drain out if we exit early
+			for range out {
+			}
+		}()
+
+		for v := range out {
+			if !yield(v, nil) {
+				return
+			}
+		}
+
+		if err != nil {
+			var z S
+			yield(z, err)
+		}
+	}
+}
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 0309063..7519c9a 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -26,15 +26,24 @@ import (
 	"time"
 
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/config"
 	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
+	tblutils "github.com/apache/iceberg-go/table/internal"
 	"github.com/google/uuid"
 	"golang.org/x/sync/errgroup"
 )
 
 type producerImpl interface {
+	// to perform any post-processing on the manifests before writing them
+	// to the new snapshot. This will be called as the last step
+	// before writing a manifest list file, using the result of this function
+	// as the final list of manifests to write.
 	processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error)
+	// perform any processing necessary and return the list of existing
+	// manifests that should be included in the snapshot
 	existingManifests() ([]iceberg.ManifestFile, error)
+	// return the deleted entries for writing delete file manifests
 	deletedEntries() ([]iceberg.ManifestEntry, error)
 }
 
@@ -87,9 +96,147 @@ func (fa *fastAppendFiles) existingManifests() ([]iceberg.ManifestFile, error) {
 }
 
 func (fa *fastAppendFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+	// for fast appends, there are no deleted entries
 	return nil, nil
 }
 
+type overwriteFiles struct {
+	base *snapshotProducer
+}
+
+func newOverwriteFilesProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+	prod := createSnapshotProducer(op, txn, fs, commitUUID, snapshotProps)
+	prod.producerImpl = &overwriteFiles{base: prod}
+
+	return prod
+}
+
+func (of *overwriteFiles) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+	// no post processing
+	return manifests, nil
+}
+
+func (of *overwriteFiles) existingManifests() ([]iceberg.ManifestFile, error) {
+	// determine if there are any existing manifest files
+	existingFiles := make([]iceberg.ManifestFile, 0)
+
+	snap := of.base.txn.meta.currentSnapshot()
+	if snap == nil {
+		return existingFiles, nil
+	}
+
+	manifestList, err := snap.Manifests(of.base.io)
+	if err != nil {
+		return existingFiles, err
+	}
+
+	for _, m := range manifestList {
+		entries, err := of.base.fetchManifestEntry(m, true)
+		if err != nil {
+			return existingFiles, err
+		}
+
+		foundDeleted := make([]iceberg.ManifestEntry, 0)
+		notDeleted := make([]iceberg.ManifestEntry, 0, len(entries))
+		for _, entry := range entries {
+			if _, ok := of.base.deletedFiles[entry.DataFile().FilePath()]; ok {
+				foundDeleted = append(foundDeleted, entry)
+			} else {
+				notDeleted = append(notDeleted, entry)
+			}
+		}
+
+		if len(foundDeleted) == 0 {
+			existingFiles = append(existingFiles, m)
+
+			continue
+		}
+
+		if len(notDeleted) == 0 {
+			continue
+		}
+
+		spec, err := of.base.txn.meta.GetSpecByID(int(m.PartitionSpecID()))
+		if err != nil {
+			return existingFiles, err
+		}
+
+		wr, path, counter, err := of.base.newManifestWriter(*spec)
+		if err != nil {
+			return existingFiles, err
+		}
+		defer counter.W.(io.Closer).Close()
+
+		for _, entry := range notDeleted {
+			if err := wr.Existing(entry); err != nil {
+				return existingFiles, err
+			}
+		}
+
+		mf, err := wr.ToManifestFile(path, counter.Count)
+		if err != nil {
+			return existingFiles, err
+		}
+
+		existingFiles = append(existingFiles, mf)
+	}
+
+	return existingFiles, nil
+}
+
+func (of *overwriteFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+	// determine if we need to record any deleted entries
+	//
+	// with a full overwrite all the entries are considered deleted
+	// with partial overwrites we have to use the predicate to evaluate
+	// which entries are affected
+	if of.base.parentSnapshotID <= 0 {
+		return nil, nil
+	}
+
+	parent, err := of.base.txn.meta.SnapshotByID(of.base.parentSnapshotID)
+	if err != nil {
+		return nil, fmt.Errorf("%w: cannot overwrite empty table", err)
+	}
+
+	previousManifests, err := parent.Manifests(of.base.io)
+	if err != nil {
+		return nil, err
+	}
+
+	getEntries := func(m iceberg.ManifestFile) ([]iceberg.ManifestEntry, error) {
+		entries, err := of.base.fetchManifestEntry(m, true)
+		if err != nil {
+			return nil, err
+		}
+
+		result := make([]iceberg.ManifestEntry, 0, len(entries))
+		for _, entry := range entries {
+			_, ok := of.base.deletedFiles[entry.DataFile().FilePath()]
+			if ok && entry.DataFile().ContentType() == iceberg.EntryContentData {
+				seqNum := entry.SequenceNum()
+				result = append(result,
+					iceberg.NewManifestEntry(iceberg.EntryStatusDELETED,
+						&of.base.snapshotID, &seqNum, entry.FileSequenceNum(),
+						entry.DataFile()))
+			}
+		}
+
+		return result, nil
+	}
+
+	nWorkers := config.EnvConfig.MaxWorkers
+	finalResult := make([]iceberg.ManifestEntry, 0, len(previousManifests))
+	for entries, err := range tblutils.MapExec(nWorkers, previousManifests, getEntries) {
+		if err != nil {
+			return nil, err
+		}
+		finalResult = append(finalResult, entries...)
+	}
+
+	return finalResult, nil
+}
+
 type snapshotProducer struct {
 	producerImpl
 
@@ -148,6 +295,30 @@ func (sp *snapshotProducer) appendDataFile(df iceberg.DataFile) *snapshotProduce
 	return sp
 }
 
+func (sp *snapshotProducer) deleteDataFile(df iceberg.DataFile) *snapshotProducer {
+	sp.deletedFiles[df.FilePath()] = df
+
+	return sp
+}
+
+func (sp *snapshotProducer) newManifestWriter(spec iceberg.PartitionSpec) (*iceberg.ManifestWriter, string, *internal.CountingWriter, error) {
+	out, path, err := sp.newManifestOutput()
+	if err != nil {
+		return nil, "", nil, err
+	}
+
+	counter := &internal.CountingWriter{W: out}
+	wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter, spec,
+		sp.txn.meta.CurrentSchema(), sp.snapshotID)
+	if err != nil {
+		defer out.Close()
+
+		return nil, "", nil, err
+	}
+
+	return wr, path, counter, nil
+}
+
 func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error) {
 	provider, err := sp.txn.tbl.LocationProvider()
 	if err != nil {
@@ -163,6 +334,10 @@ func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error)
 	return f, filepath, nil
 }
 
+func (sp *snapshotProducer) fetchManifestEntry(m iceberg.ManifestFile, discardDeleted bool) ([]iceberg.ManifestEntry, error) {
+	return m.FetchEntries(sp.io, discardDeleted)
+}
+
 func (sp *snapshotProducer) manifests() ([]iceberg.ManifestFile, error) {
 	var g errgroup.Group
 
@@ -290,7 +465,7 @@ func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
 	return updateSnapshotSummaries(Summary{
 		Operation:  sp.op,
 		Properties: summaryProps,
-	}, previousSummary, sp.op == OpOverwrite)
+	}, previousSummary)
 }
 
 func (sp *snapshotProducer) commit() ([]Update, []Requirement, error) {
diff --git a/table/snapshots.go b/table/snapshots.go
index 45b21ad..b034529 100644
--- a/table/snapshots.go
+++ b/table/snapshots.go
@@ -435,32 +435,7 @@ func (s *SnapshotSummaryCollector) build() iceberg.Properties {
 	return props
 }
 
-func truncateTableSummary(sum Summary, previous iceberg.Properties) Summary {
-	keys := []string{
-		totalDataFilesKey, totalDeleteFilesKey, totalRecordsKey,
-		totalFileSizeKey, totalPosDeletesKey, totalEqDeletesKey,
-	}
-	for _, prop := range keys {
-		sum.Properties[prop] = "0"
-	}
-
-	updateProp := func(prop, prevProp string) {
-		if val := previous.GetInt(prevProp, 0); val > 0 {
-			sum.Properties[prop] = strconv.Itoa(val)
-		}
-	}
-
-	updateProp(deletedDataFilesKey, totalDataFilesKey)
-	updateProp(removedDeleteFilesKey, totalDeleteFilesKey)
-	updateProp(deletedRecordsKey, totalRecordsKey)
-	updateProp(removedFileSizeKey, totalFileSizeKey)
-	updateProp(removedPosDeletesKey, totalPosDeletesKey)
-	updateProp(removedEqDeletesKey, totalEqDeletesKey)
-
-	return sum
-}
-
-func updateSnapshotSummaries(sum Summary, previous iceberg.Properties, truncateFullTable bool) (Summary, error) {
+func updateSnapshotSummaries(sum Summary, previous iceberg.Properties) (Summary, error) {
 	switch sum.Operation {
 	case OpAppend, OpOverwrite, OpDelete:
 	default:
@@ -471,10 +446,6 @@ func updateSnapshotSummaries(sum Summary, previous iceberg.Properties, truncateF
 		sum.Properties = make(iceberg.Properties)
 	}
 
-	if truncateFullTable && sum.Operation == OpOverwrite && previous != nil {
-		sum = truncateTableSummary(sum, previous)
-	}
-
 	if previous == nil {
 		previous = iceberg.Properties{
 			totalDataFilesKey:   "0",
diff --git a/table/transaction.go b/table/transaction.go
index c9c3c06..8b6d8cb 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -30,6 +30,7 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/io"
+	"github.com/google/uuid"
 )
 
 type snapshotUpdate struct {
@@ -42,6 +43,15 @@ func (s snapshotUpdate) fastAppend() *snapshotProducer {
 	return newFastAppendFilesProducer(OpAppend, s.txn, s.io, nil, s.snapshotProps)
 }
 
+func (s snapshotUpdate) mergeOverwrite(commitUUID *uuid.UUID) *snapshotProducer {
+	op := OpOverwrite
+	if s.txn.meta.currentSnapshot() == nil {
+		op = OpAppend
+	}
+
+	return newOverwriteFilesProducer(op, s.txn, s.io, commitUUID, s.snapshotProps)
+}
+
 type Transaction struct {
 	tbl  *Table
 	meta *MetadataBuilder
@@ -117,6 +127,101 @@ func (t *Transaction) SetProperties(props iceberg.Properties) error {
 	return nil
 }
 
+// ReplaceFiles is actually just an overwrite operation with multiple
+// files deleted and added.
+//
+// TODO: technically, this could be a REPLACE operation but we aren't performing
+// any validation here that there are no changes to the underlying data. A REPLACE
+// operation is only valid if the data is exactly the same as the previous snapshot.
+//
+// For now, we'll keep using an overwrite operation.
+func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete, filesToAdd []string, snapshotProps iceberg.Properties) error {
+	if len(filesToDelete) == 0 {
+		if len(filesToAdd) > 0 {
+			return t.AddFiles(ctx, filesToAdd, snapshotProps, false)
+		}
+	}
+
+	var (
+		setToDelete = make(map[string]struct{})
+		setToAdd    = make(map[string]struct{})
+	)
+
+	for _, f := range filesToDelete {
+		setToDelete[f] = struct{}{}
+	}
+
+	for _, f := range filesToAdd {
+		setToAdd[f] = struct{}{}
+	}
+
+	if len(setToDelete) != len(filesToDelete) {
+		return errors.New("delete file paths must be unique for ReplaceDataFiles")
+	}
+
+	if len(setToAdd) != len(filesToAdd) {
+		return errors.New("add file paths must be unique for ReplaceDataFiles")
+	}
+
+	s := t.meta.currentSnapshot()
+	if s == nil {
+		return fmt.Errorf("%w: cannot replace files in a table without an existing snapshot", ErrInvalidOperation)
+	}
+
+	markedForDeletion := make([]iceberg.DataFile, 0, len(setToDelete))
+	for df, err := range s.dataFiles(t.tbl.fs, nil) {
+		if err != nil {
+			return err
+		}
+
+		if _, ok := setToDelete[df.FilePath()]; ok {
+			markedForDeletion = append(markedForDeletion, df)
+		}
+
+		if _, ok := setToAdd[df.FilePath()]; ok {
+			return fmt.Errorf("cannot add files that are already referenced by table, files: %s", df.FilePath())
+		}
+	}
+
+	if len(markedForDeletion) != len(setToDelete) {
+		return errors.New("cannot delete files that do not belong to the table")
+	}
+
+	if t.meta.NameMapping() == nil {
+		nameMapping := t.meta.CurrentSchema().NameMapping()
+		mappingJson, err := json.Marshal(nameMapping)
+		if err != nil {
+			return err
+		}
+		err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+		if err != nil {
+			return err
+		}
+	}
+
+	commitUUID := uuid.New()
+	updater := t.updateSnapshot(snapshotProps).mergeOverwrite(&commitUUID)
+
+	for _, df := range markedForDeletion {
+		updater.deleteDataFile(df)
+	}
+
+	dataFiles := filesToDataFiles(ctx, t.tbl.fs, t.meta, slices.Values(filesToAdd))
+	for df, err := range dataFiles {
+		if err != nil {
+			return err
+		}
+		updater.appendDataFile(df)
+	}
+
+	updates, reqs, err := updater.commit()
+	if err != nil {
+		return err
+	}
+
+	return t.apply(updates, reqs)
+}
+
 func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Properties) error {
 	return iceberg.ErrNotImplemented
 }


Test patch:
diff --git a/catalog/rest/rest_integration_test.go b/catalog/rest/rest_integration_test.go
index b68a32f..c9b57bb 100644
--- a/catalog/rest/rest_integration_test.go
+++ b/catalog/rest/rest_integration_test.go
@@ -227,7 +227,7 @@ func (s *RestIntegrationSuite) TestWriteCommitTable() {
 	defer tbl.FS().Remove(pqfile)
 
 	txn := tbl.NewTransaction()
-	s.Require().NoError(txn.AddFiles([]string{pqfile}, nil, false))
+	s.Require().NoError(txn.AddFiles(s.ctx, []string{pqfile}, nil, false))
 	updated, err := txn.Commit(s.ctx)
 	s.Require().NoError(err)
 
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index c374178..0838abd 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -142,55 +142,16 @@ func TestMergeSnapshotSummaries(t *testing.T) {
 				"total-equality-deletes": "3",
 			}},
 		},
-		{
-			Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
-				"added-data-files":       "1",
-				"added-delete-files":     "2",
-				"added-equality-deletes": "3",
-				"added-files-size":       "4",
-				"added-position-deletes": "5",
-				"added-records":          "6",
-			}},
-			iceberg.Properties{
-				"total-data-files":       "1",
-				"total-delete-files":     "1",
-				"total-equality-deletes": "1",
-				"total-files-size":       "1",
-				"total-position-deletes": "1",
-				"total-records":          "1",
-			},
-			true,
-			Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
-				"added-data-files":         "1",
-				"added-delete-files":       "2",
-				"added-equality-deletes":   "3",
-				"added-files-size":         "4",
-				"added-position-deletes":   "5",
-				"added-records":            "6",
-				"total-data-files":         "1",
-				"total-records":            "6",
-				"total-delete-files":       "2",
-				"total-equality-deletes":   "3",
-				"total-files-size":         "4",
-				"total-position-deletes":   "5",
-				"deleted-data-files":       "1",
-				"removed-delete-files":     "1",
-				"deleted-records":          "1",
-				"removed-files-size":       "1",
-				"removed-position-deletes": "1",
-				"removed-equality-deletes": "1",
-			}},
-		},
 	}
 
 	for _, tt := range tests {
-		result, err := updateSnapshotSummaries(tt.sum, tt.previous, tt.truncateTable)
+		result, err := updateSnapshotSummaries(tt.sum, tt.previous)
 		require.NoError(t, err)
 		assert.Equal(t, tt.expected, result)
 	}
 }
 
 func TestInvalidOperation(t *testing.T) {
-	_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
+	_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil)
 	assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
 }
diff --git a/table/table_test.go b/table/table_test.go
index 975f40d..4d7eb88 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -735,6 +735,106 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedCurrentSnapshotIgnoreDupli
 	t.Equal([]int32{0, 0, 0}, deleted)
 }
 
+type mockedCatalog struct{}
+
+func (m *mockedCatalog) LoadTable(ctx context.Context, ident table.Identifier, props iceberg.Properties) (*table.Table, error) {
+	return nil, nil
+}
+
+func (m *mockedCatalog) CommitTable(ctx context.Context, tbl *table.Table, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
+	bldr, err := table.MetadataBuilderFromBase(tbl.Metadata())
+	if err != nil {
+		return nil, "", err
+	}
+
+	for _, u := range updates {
+		if err := u.Apply(bldr); err != nil {
+			return nil, "", err
+		}
+	}
+
+	meta, err := bldr.Build()
+	if err != nil {
+		return nil, "", err
+	}
+
+	return meta, "", nil
+}
+
+func (t *TableWritingTestSuite) TestReplaceDataFiles() {
+	fs := iceio.LocalFS{}
+
+	files := make([]string, 0)
+	for i := range 5 {
+		filePath := fmt.Sprintf("%s/replace_data_files_v%d/data-%d.parquet", t.location, t.formatVersion, i)
+		t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
+		files = append(files, filePath)
+	}
+
+	ident := table.Identifier{"default", "replace_data_files_v" + strconv.Itoa(t.formatVersion)}
+	meta, err := table.NewMetadata(t.tableSchemaPromotedTypes, iceberg.UnpartitionedSpec,
+		table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion)})
+	t.Require().NoError(err)
+
+	ctx := context.Background()
+
+	tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &mockedCatalog{})
+	for i := range 5 {
+		tx := tbl.NewTransaction()
+		t.Require().NoError(tx.AddFiles(ctx, files[i:i+1], nil, false))
+		tbl, err = tx.Commit(ctx)
+		t.Require().NoError(err)
+	}
+
+	mflist, err := tbl.CurrentSnapshot().Manifests(tbl.FS())
+	t.Require().NoError(err)
+	t.Len(mflist, 5)
+
+	// create a parquet file that is essentially as if we merged two of
+	// the data files together
+	cols := make([]arrow.Column, 0, t.arrTablePromotedTypes.NumCols())
+	for i := range int(t.arrTablePromotedTypes.NumCols()) {
+		chkd := t.arrTablePromotedTypes.Column(i).Data()
+		duplicated := arrow.NewChunked(chkd.DataType(), append(chkd.Chunks(), chkd.Chunks()...))
+		defer duplicated.Release()
+
+		col := arrow.NewColumn(t.arrSchemaPromotedTypes.Fields()[i], duplicated)
+		defer col.Release()
+
+		cols = append(cols, *col)
+	}
+
+	combined := array.NewTable(t.arrSchemaPromotedTypes, cols, -1)
+	defer combined.Release()
+
+	combinedFilePath := fmt.Sprintf("%s/replace_data_files_v%d/combined.parquet", t.location, t.formatVersion)
+	t.writeParquet(fs, combinedFilePath, combined)
+
+	tx := tbl.NewTransaction()
+	t.Require().NoError(tx.ReplaceDataFiles(ctx, files[:2], []string{combinedFilePath}, nil))
+
+	staged, err := tx.StagedTable()
+	t.Require().NoError(err)
+
+	t.Equal(&table.Summary{
+		Operation: table.OpOverwrite,
+		Properties: iceberg.Properties{
+			"added-data-files":       "1",
+			"added-files-size":       "1082",
+			"added-records":          "4",
+			"deleted-data-files":     "2",
+			"deleted-records":        "4",
+			"removed-files-size":     "2164",
+			"total-data-files":       "4",
+			"total-delete-files":     "0",
+			"total-equality-deletes": "0",
+			"total-files-size":       "4328",
+			"total-position-deletes": "0",
+			"total-records":          "10",
+		},
+	}, staged.CurrentSnapshot().Summary)
+}
+
 func TestTableWriting(t *testing.T) {
 	suite.Run(t, &TableWritingTestSuite{formatVersion: 1})
 	suite.Run(t, &TableWritingTestSuite{formatVersion: 2})