Skip to content

Commit 551f3b5

Browse files
ebyhrebyhr
committed
Add support for renaming column in Delta Lake
1 parent a400eaf commit 551f3b5

File tree

5 files changed

+276
-3
lines changed

5 files changed

+276
-3
lines changed

‎plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import io.trino.plugin.deltalake.transactionlog.AddFileEntry;
4444
import io.trino.plugin.deltalake.transactionlog.CdfFileEntry;
4545
import io.trino.plugin.deltalake.transactionlog.CommitInfoEntry;
46+
import io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode;
4647
import io.trino.plugin.deltalake.transactionlog.DeltaLakeTransactionLogEntry;
4748
import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
4849
import io.trino.plugin.deltalake.transactionlog.MetadataEntry.Format;
@@ -266,9 +267,11 @@ public class DeltaLakeMetadata
266267
LAZY_SIMPLE_SERDE_CLASS,
267268
SEQUENCEFILE_INPUT_FORMAT_CLASS,
268269
HIVE_SEQUENCEFILE_OUTPUT_FORMAT_CLASS);
270+
// Operation names in Delta Lake https://github.com/delta-io/delta/blob/master/core/src/main/scala/org/apache/spark/sql/delta/DeltaOperations.scala
269271
public static final String CREATE_TABLE_AS_OPERATION = "CREATE TABLE AS SELECT";
270272
public static final String CREATE_TABLE_OPERATION = "CREATE TABLE";
271273
public static final String ADD_COLUMN_OPERATION = "ADD COLUMNS";
274+
public static final String RENAME_COLUMN_OPERATION = "RENAME COLUMN";
272275
public static final String INSERT_OPERATION = "WRITE";
273276
public static final String MERGE_OPERATION = "MERGE";
274277
public static final String OPTIMIZE_OPERATION = "OPTIMIZE";
@@ -1145,6 +1148,78 @@ public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle
11451148
}
11461149
}
11471150

1151+
@Override
1152+
public void renameColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle, String newColumnName)
1153+
{
1154+
DeltaLakeTableHandle table = (DeltaLakeTableHandle) tableHandle;
1155+
DeltaLakeColumnHandle deltaLakeColumn = (DeltaLakeColumnHandle) columnHandle;
1156+
String sourceColumnName = deltaLakeColumn.getName();
1157+
1158+
ColumnMappingMode columnMappingMode = getColumnMappingMode(table.getMetadataEntry());
1159+
if (columnMappingMode != ColumnMappingMode.NAME && columnMappingMode != ColumnMappingMode.ID) {
1160+
throw new TrinoException(NOT_SUPPORTED, "Cannot rename column with the column mapping: " + columnMappingMode);
1161+
}
1162+
1163+
ConnectorTableMetadata tableMetadata = getTableMetadata(session, table);
1164+
long commitVersion = table.getReadVersion() + 1;
1165+
List<String> partitionColumns = getPartitionedBy(tableMetadata.getProperties()).stream()
1166+
.map(columnName -> columnName.equals(sourceColumnName) ? newColumnName : columnName)
1167+
.collect(toImmutableList());
1168+
1169+
List<DeltaLakeColumnHandle> columns = tableMetadata.getColumns().stream()
1170+
.filter(column -> !column.isHidden())
1171+
.map(column -> toColumnHandle(
1172+
column.getName().equals(sourceColumnName) ? ColumnMetadata.builderFrom(column).setName(newColumnName).build() : column,
1173+
column.getName().equals(sourceColumnName) ? newColumnName : column.getName(),
1174+
column.getType(),
1175+
partitionColumns))
1176+
.collect(toImmutableList());
1177+
Map<String, String> columnComments = getColumnComments(table.getMetadataEntry()).entrySet().stream()
1178+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1179+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1180+
Map<String, Boolean> columnsNullability = getColumnsNullability(table.getMetadataEntry()).entrySet().stream()
1181+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1182+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1183+
Map<String, Map<String, Object>> columnMetadata = getColumnsMetadata(table.getMetadataEntry()).entrySet().stream()
1184+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1185+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1186+
try {
1187+
TransactionLogWriter transactionLogWriter = transactionLogWriterFactory.newWriter(session, table.getLocation());
1188+
appendTableEntries(
1189+
commitVersion,
1190+
transactionLogWriter,
1191+
table.getMetadataEntry().getId(),
1192+
columns,
1193+
partitionColumns,
1194+
columnComments,
1195+
columnsNullability,
1196+
columnMetadata,
1197+
table.getMetadataEntry().getConfiguration(),
1198+
RENAME_COLUMN_OPERATION,
1199+
session,
1200+
Optional.ofNullable(table.getMetadataEntry().getDescription()),
1201+
getProtocolEntry(session, table.getSchemaTableName()));
1202+
transactionLogWriter.flush();
1203+
1204+
statisticsAccess.readExtendedStatistics(session, table.getLocation()).ifPresent(existingStatistics -> {
1205+
ExtendedStatistics statistics = new ExtendedStatistics(
1206+
existingStatistics.getModelVersion(),
1207+
existingStatistics.getAlreadyAnalyzedModifiedTimeMax(),
1208+
existingStatistics.getColumnStatistics().entrySet().stream()
1209+
.map(stats -> stats.getKey().equals(sourceColumnName)
1210+
? Map.entry(newColumnName, DeltaLakeColumnStatistics.create(stats.getValue().getTotalSizeInBytes(), stats.getValue().getNdvSummary()))
1211+
: stats)
1212+
.collect(toImmutableMap(Entry::getKey, Entry::getValue)),
1213+
existingStatistics.getAnalyzedColumns()
1214+
.map(analyzedColumns -> analyzedColumns.stream().map(column -> column.equals(sourceColumnName) ? newColumnName : column).collect(toImmutableSet())));
1215+
statisticsAccess.updateExtendedStatistics(session, table.getLocation(), statistics);
1216+
});
1217+
}
1218+
catch (Exception e) {
1219+
throw new TrinoException(DELTA_LAKE_BAD_WRITE, format("Unable to rename '%s' column for: %s.%s", sourceColumnName, table.getSchemaName(), table.getTableName()), e);
1220+
}
1221+
}
1222+
11481223
private void appendTableEntries(
11491224
long commitVersion,
11501225
TransactionLogWriter transactionLogWriter,

‎plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
114114
return false;
115115

116116
case SUPPORTS_DROP_COLUMN:
117-
case SUPPORTS_RENAME_COLUMN:
118117
case SUPPORTS_SET_COLUMN_TYPE:
119118
return false;
120119

@@ -362,6 +361,33 @@ public void testDropNonEmptySchemaWithTable()
362361
assertUpdate("DROP SCHEMA " + schemaName);
363362
}
364363

364+
@Override
365+
public void testRenameColumn()
366+
{
367+
// Override because the connector doesn't support renaming columns with 'none' column mapping
368+
// There are some tests in in io.trino.tests.product.deltalake.TestDeltaLakeColumnMappingMode
369+
assertThatThrownBy(super::testRenameColumn)
370+
.hasMessageContaining("Cannot rename column with the column mapping: NONE");
371+
}
372+
373+
@Override
374+
public void testAlterTableRenameColumnToLongName()
375+
{
376+
// Override because the connector doesn't support renaming columns with 'none' column mapping
377+
// There are some tests in in io.trino.tests.product.deltalake.TestDeltaLakeColumnMappingMode
378+
assertThatThrownBy(super::testAlterTableRenameColumnToLongName)
379+
.hasMessageContaining("Cannot rename column with the column mapping: NONE");
380+
}
381+
382+
@Override
383+
public void testRenameColumnName(String columnName)
384+
{
385+
// Override because the connector doesn't support renaming columns with 'none' column mapping
386+
// There are some tests in in io.trino.tests.product.deltalake.TestDeltaLakeColumnMappingMode
387+
assertThatThrownBy(() -> super.testRenameColumnName(columnName))
388+
.hasMessageContaining("Cannot rename column with the column mapping: NONE");
389+
}
390+
365391
@Override
366392
public void testCharVarcharComparison()
367393
{

‎testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/suite/suites/SuiteDeltaLakeDatabricks104.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ public List<SuiteTestRun> getTestRuns(EnvironmentConfig config)
3232
return ImmutableList.of(
3333
testOnEnvironment(EnvSinglenodeDeltaLakeDatabricks104.class)
3434
.withGroups("configured_features", "delta-lake-databricks")
35+
.withExcludedGroups("delta-lake-exclude-104")
3536
.withExcludedTests(getExcludedTests())
3637
.build());
3738
}

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/TestGroups.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ public final class TestGroups
8585
public static final String DELTA_LAKE_DATABRICKS = "delta-lake-databricks";
8686
public static final String DELTA_LAKE_EXCLUDE_73 = "delta-lake-exclude-73";
8787
public static final String DELTA_LAKE_EXCLUDE_91 = "delta-lake-exclude-91";
88+
public static final String DELTA_LAKE_EXCLUDE_104 = "delta-lake-exclude-104";
8889
public static final String DELTA_LAKE_EXCLUDE_113 = "delta-lake-exclude-113";
8990
public static final String HUDI = "hudi";
9091
public static final String PARQUET = "parquet";

0 commit comments

Comments
 (0)