Merge pull request ClickHouse#79677 from ClickHouse/fix-delta-kernel-crash

Fix potential crash in delta-kernel implementation

Author: kssenii

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

@@ -227,60 +227,63 @@ class TableSnapshot::Iterator final : public DB::IObjectIterator
         /// but instead in data files paths directory names.
         /// So we extract these values here and put into `partitions_info`.
         DB::ObjectInfoWithPartitionColumns::PartitionColumnsInfo partitions_info;
-        for (const auto & partition_column : context->partition_columns)
+        if (partition_map)
         {
-            std::string * value;
-            /// This map is empty if columnMappingMode = ''.
-            /// (E.g. empty string, which is the default mode).
-            if (context->physical_names_map.empty())
+            for (const auto & partition_column : context->partition_columns)
             {
-                value = static_cast<std::string *>(ffi::get_from_string_map(
-                    partition_map,
-                    KernelUtils::toDeltaString(partition_column),
-                    KernelUtils::allocateString));
-            }
-            else
-            {
-                /// DeltaKernel has inconsistency, getPartitionColumns returns logical column names,
-                /// while here in partition_map we would have physical columns as map keys.
-                /// This will be fixed after switching to "transform"'s.
-                auto it = context->physical_names_map.find(partition_column);
-                if (it == context->physical_names_map.end())
+                std::string * value;
+                /// This map is empty if columnMappingMode = ''.
+                /// (E.g. empty string, which is the default mode).
+                if (context->physical_names_map.empty())
                 {
-                    throw DB::Exception(
-                        DB::ErrorCodes::LOGICAL_ERROR,
-                        "Cannot find parititon column {} in physical columns map",
-                        partition_column);
+                    value = static_cast<std::string *>(ffi::get_from_string_map(
+                        partition_map,
+                        KernelUtils::toDeltaString(partition_column),
+                        KernelUtils::allocateString));
+                }
+                else
+                {
+                    /// DeltaKernel has inconsistency, getPartitionColumns returns logical column names,
+                    /// while here in partition_map we would have physical columns as map keys.
+                    /// This will be fixed after switching to "transform"'s.
+                    auto it = context->physical_names_map.find(partition_column);
+                    if (it == context->physical_names_map.end())
+                    {
+                        throw DB::Exception(
+                            DB::ErrorCodes::LOGICAL_ERROR,
+                            "Cannot find parititon column {} in physical columns map",
+                            partition_column);
+                    }
+
+                    value = static_cast<std::string *>(ffi::get_from_string_map(
+                        partition_map,
+                        KernelUtils::toDeltaString(it->second),
+                        KernelUtils::allocateString));
                 }
 
-                value = static_cast<std::string *>(ffi::get_from_string_map(
-                    partition_map,
-                    KernelUtils::toDeltaString(it->second),
-                    KernelUtils::allocateString));
-            }
-
-            SCOPE_EXIT({ delete value; });
+                SCOPE_EXIT({ delete value; });
 
-            if (value)
-            {
-                auto name_and_type = context->schema.tryGetByName(partition_column);
-                if (!name_and_type)
+                if (value)
                 {
-                    throw DB::Exception(
-                        DB::ErrorCodes::LOGICAL_ERROR,
-                        "Cannot find column `{}` in schema, there are only columns: `{}`",
-                        partition_column, fmt::join(context->schema.getNames(), ", "));
+                    auto name_and_type = context->schema.tryGetByName(partition_column);
+                    if (!name_and_type)
+                    {
+                        throw DB::Exception(
+                            DB::ErrorCodes::LOGICAL_ERROR,
+                            "Cannot find column `{}` in schema, there are only columns: `{}`",
+                            partition_column, fmt::join(context->schema.getNames(), ", "));
+                    }
+                    partitions_info.emplace_back(
+                        name_and_type.value(),
+                        DB::parseFieldFromString(*value, name_and_type->type));
                 }
-                partitions_info.emplace_back(
-                    name_and_type.value(),
-                    DB::parseFieldFromString(*value, name_and_type->type));
             }
         }
 
         LOG_TEST(
             context->log,
             "Scanned file: {}, size: {}, num records: {}, partition columns: {}",
-            full_path, size, stats->num_records, partitions_info.size());
+            full_path, size, stats ? DB::toString(stats->num_records) : "Unknown", partitions_info.size());
 
         DB::ObjectInfoPtr object;
         if (partitions_info.empty())

tests/integration/test_database_delta/test.py

@@ -77,10 +77,12 @@ def execute_multiple_spark_queries(node, queries_list, ignore_exit_code=False):
     return execute_spark_query(node, ";".join(queries_list), ignore_exit_code)
 
 
-def test_embedded_database_and_tables(started_cluster):
+@pytest.mark.parametrize("use_delta_kernel", ["1", "0"])
+def test_embedded_database_and_tables(started_cluster, use_delta_kernel):
     node1 = started_cluster.instances["node1"]
+    node1.query("drop database if exists unity_test")
     node1.query(
-        "create database unity_test engine DataLakeCatalog('http://localhost:8080/api/2.1/unity-catalog') settings warehouse = 'unity', catalog_type='unity', vended_credentials=false",
+        f"create database unity_test engine DataLakeCatalog('http://localhost:8080/api/2.1/unity-catalog') settings warehouse = 'unity', catalog_type='unity', vended_credentials=false, allow_experimental_delta_kernel_rs={use_delta_kernel}",
         settings={"allow_experimental_database_unity_catalog": "1"},
     )
     default_tables = list(