Merge pull request ClickHouse#86879 from ClickHouse/backport/25.8/86367

Backport ClickHouse#86367 to 25.8: Fix logical error during writes to DeltaLake

Author: clickhouse-gh[bot]

src/Storages/ObjectStorage/DataLakes/DeltaLake/WriteTransaction.cpp

@@ -208,12 +208,14 @@ void WriteTransaction::create()
 void WriteTransaction::validateSchema(const DB::Block & header) const
 {
     assertTransactionCreated();
-    if (write_schema.getNames() != header.getNames())
+    auto write_column_names = write_schema.getNameSet();
+    auto header_column_names = header.getNamesAndTypesList().getNameSet();
+    if (write_column_names != header_column_names)
     {
         throw DB::Exception(
             DB::ErrorCodes::LOGICAL_ERROR,
             "Header does not match write schema. Expected: {}, got: {}",
-            fmt::join(write_schema.getNames(), ", "), fmt::join(header.getNames(), ", "));
+            fmt::join(write_column_names, ", "), fmt::join(header_column_names, ", "));
     }
 }
 

tests/integration/test_storage_delta/test.py

@@ -3326,3 +3326,39 @@ def test_writes_spark_compatibility(started_cluster):
         "[Row(id=10, name='10'), Row(id=11, name='11'), Row(id=12, name='12'), Row(id=13, name='13'), Row(id=14, name='14'), Row(id=15, name='15'), Row(id=16, name='16'), Row(id=17, name='17'), Row(id=18, name='18'), Row(id=19, name='19'), Row(id=0, name='0'), Row(id=1, name='1'), Row(id=2, name='2'), Row(id=3, name='3'), Row(id=4, name='4'), Row(id=5, name='5'), Row(id=6, name='6'), Row(id=7, name='7'), Row(id=8, name='8'), Row(id=9, name='9')]"
         == str(df)
     )
+
+
+def test_write_column_order(started_cluster):
+    instance = started_cluster.instances["node1"]
+    minio_client = started_cluster.minio_client
+    bucket = started_cluster.minio_bucket
+    table_name = randomize_table_name("test_write_column_order")
+    result_file = f"{table_name}_data"
+    schema = pa.schema([("c1", pa.int32()), ("c0", pa.string())])
+    empty_arrays = [pa.array([], type=pa.int32()), pa.array([], type=pa.string())]
+    write_deltalake(
+        f"file:///{result_file}",
+        pa.Table.from_arrays(empty_arrays, schema=schema),
+        mode="overwrite",
+    )
+    LocalUploader(instance).upload_directory(f"/{result_file}/", f"/{result_file}/")
+
+    instance.query(
+        f"CREATE TABLE {table_name} (c0 String, c1 Int32) ENGINE = DeltaLakeLocal('/{result_file}') SETTINGS output_format_parquet_compression_method = 'none'"
+    )
+    num_rows = 10
+    instance.query(
+        f"INSERT INTO {table_name} (c1, c0) SELECT number as c1, toString(number % 2) as c0 FROM numbers(10)"
+    )
+
+    assert num_rows == int(instance.query(f"SELECT count() FROM {table_name}"))
+    assert (
+        "0\t0\n1\t1\n0\t2\n1\t3\n0\t4\n1\t5\n0\t6\n1\t7\n0\t8\n1\t9"
+        == instance.query(f"SELECT c0, c1 FROM {table_name}").strip()
+    )
+
+    instance.query(
+        f"INSERT INTO {table_name} (c1, c0) SELECT c1, c0 FROM generateRandom('c1 Int32, c0 String', 16920040705558589162, 7706, 3) LIMIT {num_rows}"
+    )
+
+    assert num_rows * 2 == int(instance.query(f"SELECT count() FROM {table_name}"))