Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -828,10 +828,11 @@ void IcebergMetadata::checkAlterIsPossible(const AlterCommands & commands)
ErrorCodes::NOT_IMPLEMENTED,
"Removing column property '{}' from column '{}' is not supported by Iceberg storage", command.to_remove, command.column_name);

if (command.type == AlterCommand::Type::MODIFY_COLUMN && !command.data_type)
if (command.type == AlterCommand::Type::MODIFY_COLUMN && !command.data_type
&& !command.first && command.after_column.empty())
throw Exception(
ErrorCodes::NOT_IMPLEMENTED,
"Modifying column '{}' without changing its type is not supported by Iceberg storage", command.column_name);
"Modifying column '{}' without changing its type or position is not supported by Iceberg storage", command.column_name);
}
}

Expand Down
181 changes: 137 additions & 44 deletions src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -722,7 +722,7 @@ void MetadataGenerator::generateDropColumnMetadata(const String & column_name)
metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
}

void MetadataGenerator::generateAddColumnMetadata(const String & column_name, DataTypePtr type)
void MetadataGenerator::generateAddColumnMetadata(const String & column_name, DataTypePtr type, bool first, const String & after_column)
{
if (!type->isNullable())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow to add non-nullable columns");
Expand All @@ -748,81 +748,174 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da

metadata_object->set(Iceberg::f_last_column_id, last_column_id + 1);

current_schema->getArray(Iceberg::f_fields)->add(new_field);
if (first || !after_column.empty())
{
Poco::JSON::Array::Ptr new_fields = new Poco::JSON::Array;
if (first)
{
new_fields->add(new_field);
for (UInt32 i = 0; i < existing_fields->size(); ++i)
new_fields->add(existing_fields->get(i));
}
else
{
bool inserted = false;
for (UInt32 i = 0; i < existing_fields->size(); ++i)
{
new_fields->add(existing_fields->get(i));
if (existing_fields->getObject(i)->getValue<String>(Iceberg::f_name) == after_column)
{
new_fields->add(new_field);
inserted = true;
}
}
if (!inserted)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} not found for AFTER positioning", after_column);
}
current_schema->set(Iceberg::f_fields, new_fields);
}
else
{
existing_fields->add(new_field);
}

current_schema->set(Iceberg::f_schema_id, next_schema_id);
metadata_object->set(Iceberg::f_current_schema_id, next_schema_id);
metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
}

bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name, DataTypePtr type, ContextPtr context)
bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name, DataTypePtr type, ContextPtr context, bool first, const String & after_column)
{
auto current_schema = getCurrentSchema();

auto last_column_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
auto new_type = Iceberg::getIcebergType(type, last_column_id);
auto schema_fields = current_schema->getArray(Iceberg::f_fields);

for (UInt32 i = 0; i < schema_fields->size(); ++i)
bool needs_reposition = first || !after_column.empty();
bool type_changed = false;

if (type)
{
auto current_field = schema_fields->getObject(i);
if (current_field->getValue<String>(Iceberg::f_name) == column_name)
auto new_type = Iceberg::getIcebergType(type, last_column_id);

for (UInt32 i = 0; i < schema_fields->size(); ++i)
{
auto current_field = schema_fields->getObject(i);
if (current_field->getValue<String>(Iceberg::f_name) != column_name)
continue;

if (current_field->getValue<bool>(Iceberg::f_required) == new_type.second
&& icebergTypesEqualIgnoringIds(current_field->get(Iceberg::f_type), new_type.first))
{
auto existing_iceberg_type = current_field->get(Iceberg::f_type);
if (existing_iceberg_type.isString())
if (!needs_reposition)
{
auto reconstructed_ch_type = Iceberg::IcebergSchemaProcessor::getSimpleType(
existing_iceberg_type.extract<String>(),
context,
context->getSettingsRef()[Setting::allow_experimental_geo_types_in_iceberg]);
if (!current_field->getValue<bool>(Iceberg::f_required) && reconstructed_ch_type->canBeInsideNullable())
reconstructed_ch_type = makeNullable(reconstructed_ch_type);
auto existing_iceberg_type = current_field->get(Iceberg::f_type);
if (existing_iceberg_type.isString())
{
auto reconstructed_ch_type = Iceberg::IcebergSchemaProcessor::getSimpleType(
existing_iceberg_type.extract<String>(),
context,
context->getSettingsRef()[Setting::allow_experimental_geo_types_in_iceberg]);
if (!current_field->getValue<bool>(Iceberg::f_required) && reconstructed_ch_type->canBeInsideNullable())
reconstructed_ch_type = makeNullable(reconstructed_ch_type);

if (reconstructed_ch_type->equals(*type))
return false;

if (reconstructed_ch_type->equals(*type))
return false;
throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"Cannot MODIFY COLUMN '{}' from {} to {}: both map to the same Iceberg type '{}' "
"so the change cannot be recorded in the Iceberg schema",
column_name,
reconstructed_ch_type->getName(),
type->getName(),
existing_iceberg_type.extract<String>());
}

throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"Cannot MODIFY COLUMN '{}' from {} to {}: both map to the same Iceberg type '{}' "
"so the change cannot be recorded in the Iceberg schema",
column_name,
reconstructed_ch_type->getName(),
type->getName(),
existing_iceberg_type.extract<String>());
"Cannot MODIFY COLUMN '{}': the requested and existing types both map to the same "
"Iceberg complex type, and the change cannot be recorded in the Iceberg schema",
column_name);
}

throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"Cannot MODIFY COLUMN '{}': the requested and existing types both map to the same "
"Iceberg complex type, and the change cannot be recorded in the Iceberg schema",
column_name);
}
else
{
if (!checkValidSchemaEvolution(current_field->get(Iceberg::f_type), new_type.first))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow schema evolution to type {}", type->getPrettyName());

if (!checkValidSchemaEvolution(current_field->get(Iceberg::f_type), new_type.first))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow schema evolution to type {}", type->getPrettyName());
if (!current_field->getValue<bool>(Iceberg::f_required) && !type->isNullable())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow change type from nullable to non-nullable {}", type->getPrettyName());

if (!current_field->getValue<bool>(Iceberg::f_required) && !type->isNullable())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow change type from nullable to non-nullable {}", type->getPrettyName());
type_changed = true;
}
break;
}
}

UInt32 target_index = static_cast<UInt32>(schema_fields->size());
for (UInt32 i = 0; i < schema_fields->size(); ++i)
{
if (schema_fields->getObject(i)->getValue<String>(Iceberg::f_name) == column_name)
{
target_index = i;
break;
}
}
if (target_index == schema_fields->size())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} not found in schema", column_name);

const auto next_schema_id = getNextSchemaId(metadata_object);
if (!type_changed && !needs_reposition)
return false;

current_schema = deepCopy(current_schema);
schema_fields = current_schema->getArray(Iceberg::f_fields);
current_field = schema_fields->getObject(i);
const auto next_schema_id = getNextSchemaId(metadata_object);
current_schema = deepCopy(current_schema);
schema_fields = current_schema->getArray(Iceberg::f_fields);
auto target_field = schema_fields->getObject(target_index);

current_field->set(Iceberg::f_type, new_type.first);
current_field->set(Iceberg::f_required, new_type.second);
if (type_changed)
{
auto new_type = Iceberg::getIcebergType(type, last_column_id);
target_field->set(Iceberg::f_type, new_type.first);
target_field->set(Iceberg::f_required, new_type.second);
}

metadata_object->set(Iceberg::f_current_schema_id, next_schema_id);
current_schema->set(Iceberg::f_schema_id, next_schema_id);
metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
return true;
if (needs_reposition)
{
Poco::JSON::Array::Ptr new_fields = new Poco::JSON::Array;
if (first)
{
new_fields->add(schema_fields->get(target_index));
for (UInt32 i = 0; i < schema_fields->size(); ++i)
{
if (i != target_index)
new_fields->add(schema_fields->get(i));
}
}
else
{
bool inserted = false;
for (UInt32 i = 0; i < schema_fields->size(); ++i)
{
if (i == target_index)
continue;
new_fields->add(schema_fields->get(i));
if (schema_fields->getObject(i)->getValue<String>(Iceberg::f_name) == after_column)
{
new_fields->add(schema_fields->get(target_index));
inserted = true;
}
}
if (!inserted)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} not found for AFTER positioning", after_column);
}
current_schema->set(Iceberg::f_fields, new_fields);
}

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} not found in schema", column_name);
metadata_object->set(Iceberg::f_current_schema_id, next_schema_id);
current_schema->set(Iceberg::f_schema_id, next_schema_id);
metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
return true;
}

void MetadataGenerator::generateRenameColumnMetadata(const String & column_name, const String & new_column_name)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,18 +41,18 @@ class MetadataGenerator
std::optional<Int64> user_defined_timestamp = std::nullopt,
bool is_truncate = false);

void generateAddColumnMetadata(const String & column_name, DataTypePtr type, bool first = false, const String & after_column = {});
/// Create a manifest-only rewrite snapshot (`replace` operation) carrying `total-*` counters forward so `OPTIMIZE ... MANIFEST` is idempotent.
NextMetadataResult generateManifestOnlySnapshot(
FileNamesGenerator & generator,
const Iceberg::IcebergPathFromMetadata & metadata_file_path,
Int64 parent_snapshot_id);

void generateAddColumnMetadata(const String & column_name, DataTypePtr type);
void generateDropColumnMetadata(const String & column_name);
/// Returns false when the column already has the requested type (no metadata change).
/// Returns false when neither the type nor the position changed (true no-op).
/// `context` supplies the settings used to map the stored Iceberg type back to a ClickHouse
/// type (the timestamptz timezone and whether geo types are allowed).
bool generateModifyColumnMetadata(const String & column_name, DataTypePtr type, ContextPtr context);
bool generateModifyColumnMetadata(const String & column_name, DataTypePtr type, ContextPtr context, bool first = false, const String & after_column = {});
void generateRenameColumnMetadata(const String & column_name, const String & new_column_name);

/// A commit attempt can land in the catalog even when the client observes a failure
Expand Down
4 changes: 2 additions & 2 deletions src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -900,7 +900,7 @@ void alter(
switch (params[0].type)
{
case AlterCommand::Type::ADD_COLUMN:
metadata_json_generator.generateAddColumnMetadata(params[0].column_name, params[0].data_type);
metadata_json_generator.generateAddColumnMetadata(params[0].column_name, params[0].data_type, params[0].first, params[0].after_column);
break;
case AlterCommand::Type::DROP_COLUMN:
if (params[0].clear)
Expand All @@ -909,7 +909,7 @@ void alter(
break;
case AlterCommand::Type::MODIFY_COLUMN:
{
if (!metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type, context))
if (!metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type, context, params[0].first, params[0].after_column))
{
succeeded = true;
}
Expand Down
Loading
Loading