Fix iceberg tests

Author: kssenii

src/Storages/ObjectStorage/StorageObjectStorageSource.cpp

@@ -550,17 +550,24 @@ StorageObjectStorageSource::ReaderHolder StorageObjectStorageSource::createReade
 
         builder.init(Pipe(input_format));
 
-        std::shared_ptr<const ActionsDAG> transformer;
-        if (object_info->data_lake_metadata)
-            transformer = object_info->data_lake_metadata->transform;
+        std::optional<ActionsDAG> transformer;
+        if (object_info->data_lake_metadata && object_info->data_lake_metadata->transform)
+        {
+            transformer = object_info->data_lake_metadata->transform->clone();
+            /// FIXME: This is currently not done for the below case (configuration->getSchemaTransformer())
+            /// because it is an iceberg case where transformer contains columns ids (just increasing numbers)
+            /// which do not match requested_columns (while here requested_columns were adjusted to match physical columns).
+            transformer->removeUnusedActions(read_from_format_info.requested_columns.getNames());
+        }
         if (!transformer)
-            transformer = configuration->getSchemaTransformer(context_, object_info->getPath());
+        {
+            if (auto schema_transformer = configuration->getSchemaTransformer(context_, object_info->getPath()))
+                transformer = schema_transformer->clone();
+        }
 
-        if (transformer)
+        if (transformer.has_value())
         {
-            auto transformer_copy = transformer->clone();
-            transformer_copy.removeUnusedActions(read_from_format_info.requested_columns.getNames());
-            auto schema_modifying_actions = std::make_shared<ExpressionActions>(std::move(transformer_copy));
+            auto schema_modifying_actions = std::make_shared<ExpressionActions>(std::move(transformer.value()));
             builder.addSimpleTransform([&](const SharedHeader & header)
             {
                 return std::make_shared<ExpressionTransform>(header, schema_modifying_actions);

tests/integration/test_storage_delta/test.py

@@ -2229,10 +2229,10 @@ def test_column_pruning(started_cluster):
         ]
     )
 
-    num_rows = 100000
+    num_rows = 10000
     now = datetime.now()
     data = [
-        (i, f"name_{i}", 32, "".join("a" for _ in range(1000)), "2025")
+        (i, f"name_{i}", 32, "".join("a" for _ in range(100)), "2025")
         for i in range(num_rows)
     ]
     df = spark.createDataFrame(data=data, schema=schema)
@@ -2258,7 +2258,7 @@ def test_column_pruning(started_cluster):
         )
     )
     instance.query("SYSTEM FLUSH LOGS")
-    assert 467413 == int(
+    assert 107220 == int(
         instance.query(
             f"SELECT ProfileEvents['ReadBufferFromS3Bytes'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
         )
@@ -2278,7 +2278,7 @@ def test_column_pruning(started_cluster):
         )
     )
     # Small diff because in case of delta-kernel metadata reading is not counted in the metric.
-    assert 465864 == int(
+    assert 105677 == int(
         instance.query(
             f"SELECT ProfileEvents['ReadBufferFromS3Bytes'] FROM system.query_log WHERE query_id = '{query_id}' and type = 'QueryFinish'"
         )