Fix min/max value in Iceberg writes

ianton-ru · ianton-ru · commit f70f7a9dce71 · 2025-12-05T16:53:12.000+01:00
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp
@@ -1232,11 +1232,6 @@ void IcebergStorageSink::consume(Chunk & chunk)
         {
             auto [data_filename, data_filename_in_storage] = filename_generator.generateDataFileName();
             data_filenames[partition_key] = data_filename;
-            if (!statistics.contains(partition_key))
-            {
-                statistics.emplace(partition_key, current_schema->getArray(Iceberg::f_fields));
-            }
-            statistics.at(partition_key).update(part_chunk);
 
             auto buffer = object_storage->writeObject(
                 StoredObject(data_filename_in_storage), WriteMode::Rewrite, std::nullopt, DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings());
@@ -1252,6 +1247,12 @@ void IcebergStorageSink::consume(Chunk & chunk)
                 configuration->format, *write_buffers[partition_key], *sample_block, context, format_settings);
         }
 
+        if (!statistics.contains(partition_key))
+        {
+            statistics.emplace(partition_key, current_schema->getArray(Iceberg::f_fields));
+        }
+        statistics.at(partition_key).update(part_chunk);
+
         writers[partition_key]->write(getHeader().cloneWithColumns(part_chunk.getColumns()));
     }
 }
diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py
@@ -3269,3 +3269,18 @@ def check_validity_and_get_prunned_files(select_expression):
         )
         == 3
     )
+
+
+def test_iceberg_write_minmax(started_cluster):
+    instance = started_cluster.instances["node1"]
+    TABLE_NAME = "test_iceberg_write_minmax_" + get_uuid_str()
+
+    create_iceberg_table("local", instance, TABLE_NAME, started_cluster, "(x Int32, y Int32)", partition_by="identity(x)")
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 1), (1, 2)", settings={"allow_experimental_insert_into_iceberg": 1})
+
+    res = instance.query(f"SELECT x,y FROM {TABLE_NAME} WHERE y=1 ORDER BY ALL").strip()
+    assert res == "1\t1"
+
+    res = instance.query(f"SELECT x,y FROM {TABLE_NAME} WHERE y=2 ORDER BY ALL").strip()
+    assert res == "1\t2"

Original file line number	Diff line number	Diff line change
`@@ -1232,11 +1232,6 @@ void IcebergStorageSink::consume(Chunk & chunk)`
`1232`	`1232`	`{`
`1233`	`1233`	`auto [data_filename, data_filename_in_storage] = filename_generator.generateDataFileName();`
`1234`	`1234`	`data_filenames[partition_key] = data_filename;`
`1235`		`- if (!statistics.contains(partition_key))`
`1236`		`- {`
`1237`		`- statistics.emplace(partition_key, current_schema->getArray(Iceberg::f_fields));`
`1238`		`- }`
`1239`		`- statistics.at(partition_key).update(part_chunk);`
`1240`	`1235`
`1241`	`1236`	`auto buffer = object_storage->writeObject(`
`1242`	`1237`	`StoredObject(data_filename_in_storage), WriteMode::Rewrite, std::nullopt, DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings());`
`@@ -1252,6 +1247,12 @@ void IcebergStorageSink::consume(Chunk & chunk)`
`1252`	`1247`	`configuration->format, write_buffers[partition_key], sample_block, context, format_settings);`
`1253`	`1248`	`}`
`1254`	`1249`
	`1250`	`+ if (!statistics.contains(partition_key))`
	`1251`	`+ {`
	`1252`	`+ statistics.emplace(partition_key, current_schema->getArray(Iceberg::f_fields));`
	`1253`	`+ }`
	`1254`	`+ statistics.at(partition_key).update(part_chunk);`
	`1255`	`+`
`1255`	`1256`	`writers[partition_key]->write(getHeader().cloneWithColumns(part_chunk.getColumns()));`
`1256`	`1257`	`}`
`1257`	`1258`	`}`