Create symlinks using table's actual data path for Atomic db

src/Databases/DatabaseAtomic.cpp

@@ -210,7 +210,7 @@ void DatabaseAtomic::renameTable(ContextPtr local_context, const String & table_
             return;
         db.table_name_to_path.emplace(table_name_, table_data_path_);
         if (table_->storesDataOnDisk())
-            db.tryCreateSymlink(table_name_, table_data_path_);
+            db.tryCreateSymlink(table_);
     };
 
     auto assert_can_move_mat_view = [inside_database](const StoragePtr & table_)
@@ -336,7 +336,7 @@ void DatabaseAtomic::commitCreateTable(const ASTCreateQuery & query, const Stora
         throw;
     }
     if (table->storesDataOnDisk())
-        tryCreateSymlink(query.getTable(), table_data_path);
+        tryCreateSymlink(table);
 }
 
 void DatabaseAtomic::commitAlterTable(const StorageID & table_id, const String & table_metadata_tmp_path, const String & table_metadata_path,
@@ -467,7 +467,14 @@ LoadTaskPtr DatabaseAtomic::startupDatabaseAsync(AsyncLoader & async_loader, Loa
 
             fs::create_directories(path_to_table_symlinks);
             for (const auto & table : table_names)
-                tryCreateSymlink(table.first, table.second, true);
+            {
+                /// All tables in database should be loaded at this point
+                StoragePtr table_ptr = tryGetTable(table.first, getContext());
+                if (table_ptr)
+                    tryCreateSymlink(table_ptr, true);
+                else
+                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Table {} is not loaded before database startup", table.first);
+            }
         });
     std::scoped_lock lock(mutex);
     return startup_atomic_database_task = makeLoadTask(async_loader, {job});
@@ -495,12 +502,17 @@ void DatabaseAtomic::stopLoading()
     DatabaseOrdinary::stopLoading();
 }
 
-void DatabaseAtomic::tryCreateSymlink(const String & table_name, const String & actual_data_path, bool if_data_path_exist)
+void DatabaseAtomic::tryCreateSymlink(const StoragePtr & table, bool if_data_path_exist)
 {
     try
     {
+        String table_name = table->getStorageID().getTableName();
+
+        if (!table->storesDataOnDisk())
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Table {} doesn't have data path to create symlink", table_name);
+
         String link = path_to_table_symlinks + escapeForFileName(table_name);
-        fs::path data = fs::canonical(getContext()->getPath()) / actual_data_path;
+        fs::path data = fs::canonical(table->getDataPaths()[0]);
 
         /// If it already points where needed.
         std::error_code ec;

src/Databases/DatabaseAtomic.h

@@ -59,7 +59,7 @@ class DatabaseAtomic : public DatabaseOrdinary
 
     UUID tryGetTableUUID(const String & table_name) const override;
 
-    void tryCreateSymlink(const String & table_name, const String & actual_data_path, bool if_data_path_exist = false);
+    void tryCreateSymlink(const StoragePtr & table, bool if_data_path_exist = false);
     void tryRemoveSymlink(const String & table_name);
 
     void waitDetachedTableNotInUse(const UUID & uuid) override;

tests/integration/test_filesystem_layout/configs/config.d/storage_configuration.xml

@@ -0,0 +1,38 @@
+<clickhouse>
+    <storage_configuration>
+        <disks>
+            <default>
+                <keep_free_space_bytes>1024</keep_free_space_bytes>
+            </default>
+            <jbod1>
+                <path>/jbod1/</path>
+            </jbod1>
+            <jbod2>
+                <path>/jbod2/</path>
+            </jbod2>
+            <s3>
+                <type>s3</type>
+                <endpoint>http://minio1:9001/root/data/</endpoint>
+                <access_key_id>minio</access_key_id>
+                <secret_access_key>minio123</secret_access_key>
+            </s3>
+        </disks>
+        <policies>
+            <jbod>
+                <volumes>
+                    <jbod_volume>
+                        <disk>jbod1</disk>
+                        <disk>jbod2</disk>
+                    </jbod_volume>
+                </volumes>
+            </jbod>
+            <s3>
+                <volumes>
+                    <s3_volume>
+                        <disk>s3</disk>
+                    </s3_volume>
+                </volumes>
+            </s3>
+        </policies>
+    </storage_configuration>
+</clickhouse>

tests/integration/test_filesystem_layout/test.py

@@ -1,9 +1,17 @@
 import pytest
 
 from helpers.cluster import ClickHouseCluster
+from pathlib import Path
 
 cluster = ClickHouseCluster(__file__)
-node = cluster.add_instance("node")
+node = cluster.add_instance(
+    "node",
+    main_configs=[
+        "configs/config.d/storage_configuration.xml",
+    ],
+    with_minio=True,
+    stay_alive=True,
+)
 
 
 @pytest.fixture(scope="module")
@@ -79,3 +87,78 @@ def test_file_path_escaping(started_cluster):
             "test -f /var/lib/clickhouse/shadow/2/store/123/12345678-1000-4000-8000-000000000001/1_1_1_0/%7EId.bin",
         ]
     )
+
+
+def test_data_directory_symlinks(started_cluster):
+    node.query("CREATE DATABASE IF NOT EXISTS test_symlinks ENGINE = Atomic")
+
+    node.query(
+        sql="""
+            CREATE TABLE test_symlinks.default UUID '87654321-1000-4000-8000-000000000001'
+            (Id UInt32)
+            ENGINE = MergeTree()
+            PARTITION BY Id
+            ORDER BY Id
+            """,
+        database="test_symlinks",
+    )
+
+    node.query(
+        """
+        CREATE TABLE test_symlinks.s3 UUID '87654321-1000-4000-8000-000000000002'
+        (Id UInt32)
+        ENGINE = MergeTree()
+        PARTITION BY Id
+        ORDER BY Id
+        SETTINGS storage_policy = 's3'
+        """
+    )
+
+    node.query(
+        """
+        CREATE TABLE test_symlinks.jbod UUID '87654321-1000-4000-8000-000000000003'
+        (Id UInt32)
+        ENGINE = MergeTree()
+        PARTITION BY Id
+        ORDER BY Id
+        SETTINGS storage_policy = 'jbod'
+        """
+    )
+
+    clickhouse_dir = Path("/var/lib/clickhouse/")
+
+    database_dir = clickhouse_dir / "data" / "test_symlinks"
+    default_symlink = database_dir / "default"
+    s3_symlink = database_dir / "s3"
+    jbod_symlink = database_dir / "jbod"
+
+    default_data = (
+        clickhouse_dir / "store" / "876" / "87654321-1000-4000-8000-000000000001"
+    )
+    s3_data = (
+        clickhouse_dir
+        / "disks"
+        / "s3"
+        / "store"
+        / "876"
+        / "87654321-1000-4000-8000-000000000002"
+    )
+    jbod_data = Path("jbod1") / "store" / "876" / "87654321-1000-4000-8000-000000000003"
+
+    node.restart_clickhouse()
+
+    assert (
+        node.exec_in_container(["bash", "-c", f"ls -l {default_symlink}"])
+        .strip()
+        .endswith(f"{default_symlink} -> {default_data}")
+    )
+    assert (
+        node.exec_in_container(["bash", "-c", f"ls -l {s3_symlink}"])
+        .strip()
+        .endswith(f"{s3_symlink} -> {s3_data}")
+    )
+    assert (
+        node.exec_in_container(["bash", "-c", f"ls -l {jbod_symlink}"])
+        .strip()
+        .endswith(f"{jbod_symlink} -> /{jbod_data}")
+    )