const StorageID & table_id_,

ContextPtr context_,

const String & target_database_,

const String & target_table_)

: IStorage(table_id_)

, WithContext(context_->getGlobalContext())

, target_database(target_database_)

, target_table(target_table_)

{

StorageID target_id(target_database, target_table);

if (table_id_ == target_id)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Alias table cannot refer to itself");

// Disallow target is also an alias

auto target_storage = DatabaseCatalog::instance().tryGetTable(target_id, context_);

if (target_storage && target_storage->getName() == "Alias")

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Alias table cannot refer to another Alias table");

{

if (access_check)

{

if (access_check->column_names.empty())

access_check->context->checkAccess(access_check->access_type, target_database, target_table);

else

access_check->context->checkAccess(access_check->access_type, target_database, target_table, access_check->column_names);

}

return DatabaseCatalog::instance().getTable(StorageID(target_database, target_table), getContext());

{

if (!query_context)

return false;

auto access = query_context->getAccess();

if (column_name.empty())

return access->isGranted(access_type, target_database, target_table);

return access->isGranted(access_type, target_database, target_table, column_name);

{

AliasSink(

StorageAlias & storage_,

ContextPtr context_,

const StorageMetadataPtr & metadata_snapshot_,

bool async_insert_)

: SinkToStorage(std::make_shared<const Block>(metadata_snapshot_->getSampleBlock()))

, WithContext(context_)

, storage(storage_)

, non_materialized_header(metadata_snapshot_->getSampleBlockNonMaterialized())

, async_insert(async_insert_)

{

}

~AliasSink() override

{

/// On cancellation without an exception (e.g. timeout_overflow_mode='break') neither

/// onFinish() nor onException() runs, leaving the nested executor started but unfinished.

/// Cancel it so ~PushingPipelineExecutor's finished-or-unwinding invariant holds.

if (executor)

{

try

{

executor->cancel();

}

catch (...)

{

tryLogCurrentException("AliasSink");

}

}

}

String getName() const override { return "AliasSink"; }

void onStart() override

{

StoragePtr target = storage.getTargetTable();

std::unique_ptr<ASTInsertQuery> insert = std::make_unique<ASTInsertQuery>();

insert->table_id = target->getStorageID();

ASTPtr query_ptr(insert.release());

auto insert_context = Context::createCopy(getContext());

insert_context->makeQueryContext();

if (getContext()->hasQueryContext())

insert_context->setQueryAccessInfo(getContext()->getQueryContext()->getQueryAccessInfoPtr());

addInterpreterContext(insert_context);

/// This sink is one branch of the outer query's `max_insert_threads` fan-out (or its only

/// stream). Keep the nested INSERT single-stream: with the outer fan-out already in place,

/// letting every branch fan out again would multiply the number of real sink branches (part

/// writers, squashing and compression buffers) up to `max_insert_threads^2`, exceeding the

/// budget the user allowed for this INSERT.

insert_context->setSetting("max_insert_threads", 1);

/// Thread the outer async-insert flag into the nested target pipeline so INSERT through

/// Alias matches a direct insert: async batches select async dedup settings and skip the

/// strict block-limit squashing that would slice their multi-token DeduplicationInfo and

/// break its row/offset invariant.

InterpreterInsertQuery interpreter(

query_ptr,

insert_context,

/* allow_materialized */ false,

/* no_squash */ false,

/* no_destination */ false,

/* async_insert */ async_insert);

block_io = interpreter.execute();

executor = std::make_unique<PushingPipelineExecutor>(block_io.pipeline);

executor->start();

}

void consume(Chunk & chunk) override

{

if (chunk.getNumRows() == 0)

return;

auto block = getHeader().cloneWithColumns(chunk.getColumns());

Block non_materialized_block;

for (const auto & col : non_materialized_header)

non_materialized_block.insert(block.getByName(col.name));

Chunk non_materialized_chunk(non_materialized_block.getColumns(), non_materialized_block.rows());

non_materialized_chunk.setChunkInfos(chunk.getChunkInfos().clone());

/// The nested INSERT re-anchors the deduplication info to its own chunks (its squashing and

/// `AddDeduplicationInfoTransform` call `updateOriginalBlock`). When this sink is fed by a

/// dependent materialized view whose inner query changed the number of rows, those chunks

/// no longer match the rows the info's offsets describe, and computing a data hash after

/// that re-anchoring would read out of the block's bounds. Cache the hashes now, while the

/// info is still consistent.

if (auto deduplication_info = non_materialized_chunk.getChunkInfos().get<DeduplicationInfo>())

deduplication_info->cacheDataHashes(data_hash_cache);

executor->push(std::move(non_materialized_chunk));

}

void onFinish() override

{

executor->finish();

executor.reset();

block_io.onFinish();

block_io = {};

}

void onException(std::exception_ptr) override

{

if (executor)

executor->cancel();

executor.reset();

block_io.onException();

block_io = {};

}

StorageAlias & storage;

Block non_materialized_header;

bool async_insert;

BlockIO block_io;

std::unique_ptr<PushingPipelineExecutor> executor;

/// Memoizes the deduplication data hashes across the sibling chunks of one source block, so a

/// row-count-changing view fanned out into many chunks does not re-hash the source per chunk.

DeduplicationInfo::DataHashCache data_hash_cache;

QueryPlan & query_plan,

const Names & column_names,

const StorageSnapshotPtr & /*storage_snapshot*/,

SelectQueryInfo & query_info,

ContextPtr local_context,

QueryProcessingStage::Enum processed_stage,

size_t max_block_size,

size_t num_streams)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::SELECT, column_names});

auto lock = target_storage->lockForShare(

local_context->getCurrentQueryId(),

local_context->getSettingsRef()[Setting::lock_acquire_timeout]);

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

auto target_snapshot = target_storage->getStorageSnapshot(target_metadata, local_context);

target_storage->read(

query_plan,

column_names,

target_snapshot,

query_info,

local_context,

processed_stage,

max_block_size,

num_streams);

query_plan.addStorageHolder(target_storage);

query_plan.addTableLock(std::move(lock));

const ASTPtr & /*query*/,

const StorageMetadataPtr & /*metadata_snapshot*/,

ContextPtr local_context,

bool async_insert)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::INSERT});

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

/// Use AliasSink which executes full INSERT pipeline on target

/// Therefore it will trigger the MV on the target

return std::make_shared<AliasSink>(*this, local_context, target_metadata, async_insert);

const AlterCommands & params,

ContextPtr local_context,

AlterLockHolder & table_lock_holder)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::ALTER});

/// ALTER through alias on a table in a Replicated database is not supported

/// when the alias and target are in different databases. This is because the

/// DDL worker path is bypassed and metadata changes won't be replicated to

/// other replicas in ZooKeeper. If both are in the same Replicated database,

/// the DDL worker handles the ALTER correctly.

auto target_storage_id = target_storage->getStorageID();

if (getStorageID().database_name != target_storage_id.database_name)

{

auto target_db = DatabaseCatalog::instance().tryGetDatabase(target_storage_id.database_name);

if (target_db && target_db->getEngineName() == "Replicated")

{

throw Exception(

ErrorCodes::NOT_IMPLEMENTED,

"ALTER through alias is not supported when the target table is in a different Replicated database. "

"Execute the ALTER directly on the target table: {}",

target_storage_id.getNameForLogs());

}

}

target_storage->alter(params, local_context, table_lock_holder);

const ASTPtr & query,

const StorageMetadataPtr & /*metadata_snapshot*/,

ContextPtr local_context,

TableExclusiveLockHolder & table_lock_holder)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::TRUNCATE});

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

target_storage->truncate(query, target_metadata, local_context, table_lock_holder);

const ASTPtr & query,

const StorageMetadataPtr & /*metadata_snapshot*/,

const ASTPtr & partition,

bool final,

bool deduplicate,

const Names & deduplicate_by_columns,

bool cleanup,

ContextPtr local_context)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::OPTIMIZE});

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

return target_storage->optimize(query, target_metadata, partition, final, deduplicate,

deduplicate_by_columns, cleanup, local_context);

const StorageMetadataPtr &,

const PartitionCommands & commands,

ContextPtr local_context)

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::ALTER});

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

return target_storage->alterPartition(target_metadata, commands, local_context);

const PartitionCommands & commands,

const StorageMetadataPtr & /*metadata_snapshot*/,

const Settings & settings,

ContextPtr local_context) const

{

auto target_storage = getTargetTable();

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

target_storage->checkAlterPartitionIsPossible(commands, target_metadata, settings, local_context);

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::ALTER});

target_storage->mutate(commands, local_context);

{

auto target_storage = getTargetTable(TargetAccess{local_context, AccessType::ALTER});

auto lock = target_storage->lockForShare(

local_context->getCurrentQueryId(),

local_context->getSettingsRef()[Setting::lock_acquire_timeout]);

auto pipeline = target_storage->updateLightweight(commands, local_context);

/// The caller locks the alias, not the target, so the target needs its own share lock held

/// until the pipeline has committed the patch part.

QueryPlanResourceHolder target_resources;

target_resources.storage_holders.emplace_back(target_storage);

target_resources.table_locks.emplace_back(std::move(lock));

pipeline.addResources(std::move(target_resources));

return pipeline;

{

StoragePtr target;

IStorage::DataValidationTasksPtr inner;

AliasCheckTasks(StoragePtr target_, IStorage::DataValidationTasksPtr inner_)

: target(std::move(target_)), inner(std::move(inner_))

{

chassert(inner);

}

size_t size() const override { return inner->size(); }

{

auto target = getTargetTable(TargetAccess{query_context, AccessType::CHECK});

auto inner = target->getCheckTaskList(filter, query_context);

return std::make_shared<AliasCheckTasks>(std::move(target), std::move(inner));

{

auto * tasks = assert_cast<AliasCheckTasks *>(check_task_list.get());

return tasks->target->checkDataNext(tasks->inner);

{

if (!storage_snapshot)

return false;

bool has_select_access = false;

for (const auto & column : storage_snapshot->metadata->getColumns())

{

if (isTargetTableGranted(query_context, AccessType::SELECT, column.name))

{

has_select_access = true;

break;

}

}

if (!has_select_access)

return false;

auto target = tryGetTargetTable();

return target && target->supportsTrivialCountOptimization(storage_snapshot, query_context);

{

if (!isTargetTableGranted(query_context, AccessType::SHOW_TABLES, {}))

return {};

auto target = tryGetTargetTable();

return target ? target->totalRows(query_context) : std::optional<UInt64>{};

{

if (!isTargetTableGranted(query_context, AccessType::SHOW_TABLES, {}))

return {};

auto target = tryGetTargetTable();

return target ? target->totalBytes(query_context) : std::optional<UInt64>{};

{

// Only rename the alias itself, not the target table

renameInMemory(new_table_id);

ContextPtr local_context,

QueryProcessingStage::Enum to_stage,

const StorageSnapshotPtr & /* storage_snapshot */,

SelectQueryInfo & query_info) const

{

auto target_storage = getTargetTable();

auto target_metadata = target_storage->getInMemoryMetadataPtr(local_context, false);

auto target_snapshot = target_storage->getStorageSnapshot(target_metadata, local_context);

return target_storage->getQueryProcessingStage(local_context, to_stage, target_snapshot, query_info);

{

factory.registerStorage("Alias", [](const StorageFactory::Arguments & args)

{

// Supported syntaxes:

// CREATE TABLE t2 ENGINE = Alias(t)

// CREATE TABLE t2 ENGINE = Alias(db, t)

// CREATE TABLE t2 ENGINE = Alias('t')

// CREATE TABLE t2 ENGINE = Alias('db', 't')

auto local_context = args.getLocalContext();

String target_database;

String target_table;

if (args.engine_args.empty())

{

// Note: CREATE TABLE ... AS ... syntax is not supported

throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,

"Storage Alias requires target table name in ENGINE arguments. "

"Use: ENGINE = Alias('table_name') or ENGINE = Alias('database', 'table_name').");

}

else if (args.engine_args.size() == 1)

{

// Syntax: ENGINE = Alias(table_name) or ENGINE = Alias(db.table_name)

args.engine_args[0] = evaluateConstantExpressionOrIdentifierAsLiteral(args.engine_args[0], local_context);

String table_arg = checkAndGetLiteralArgument<String>(args.engine_args[0], "table_name");

auto dot_pos = table_arg.find('.');

if (dot_pos != String::npos)

{

target_database = table_arg.substr(0, dot_pos);

target_table = table_arg.substr(dot_pos + 1);

}

else

{

target_table = table_arg;

target_database = args.table_id.database_name;

}

}

else if (args.engine_args.size() == 2)

{

// Syntax: ENGINE = Alias(database_name, table_name)

args.engine_args[0] = evaluateConstantExpressionOrIdentifierAsLiteral(args.engine_args[0], local_context);

args.engine_args[1] = evaluateConstantExpressionOrIdentifierAsLiteral(args.engine_args[1], local_context);

target_database = checkAndGetLiteralArgument<String>(args.engine_args[0], "database_name");

target_table = checkAndGetLiteralArgument<String>(args.engine_args[1], "table_name");

}

else

{

throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,

"Storage Alias requires at most 2 arguments: database name and table name");

}

// Storage Alias does not support explicit column definitions

// Columns are always dynamically fetched from the target table

// Only check for CREATE, not for ATTACH/RESTORE

if (!args.columns.empty() && args.mode == LoadingStrictnessLevel::CREATE)

{

throw Exception(ErrorCodes::BAD_ARGUMENTS,

"Storage Alias does not support explicit column definitions");

}

if (!(isLoadingFromExistingMetadata(args.mode) || args.query.attach_short_syntax))

local_context->checkAccess(AccessType::SHOW_COLUMNS, target_database, target_table);

return std::make_shared<StorageAlias>(

args.table_id,

local_context,

target_database,

target_table);

},

{

.supports_schema_inference = true

},

Documentation{

.description = R"DOCS_MD(

)DOCS_MD",

.syntax = "ENGINE = Alias([target_db, ]target_table)"});