Skip to content

Commit e7d6629

Browse files
ebyhrebyhr
committed
Add support for renaming column in Delta Lake
1 parent 8cfa24c commit e7d6629

File tree

5 files changed

+278
-3
lines changed

5 files changed

+278
-3
lines changed

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

Lines changed: 77 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;
@@ -272,9 +273,11 @@ public class DeltaLakeMetadata
272273
LAZY_SIMPLE_SERDE_CLASS,
273274
SEQUENCEFILE_INPUT_FORMAT_CLASS,
274275
HIVE_SEQUENCEFILE_OUTPUT_FORMAT_CLASS);
276+
// Operation names in Delta Lake https://github.com/delta-io/delta/blob/master/core/src/main/scala/org/apache/spark/sql/delta/DeltaOperations.scala
275277
public static final String CREATE_TABLE_AS_OPERATION = "CREATE TABLE AS SELECT";
276278
public static final String CREATE_TABLE_OPERATION = "CREATE TABLE";
277279
public static final String ADD_COLUMN_OPERATION = "ADD COLUMNS";
280+
public static final String RENAME_COLUMN_OPERATION = "RENAME COLUMN";
278281
public static final String INSERT_OPERATION = "WRITE";
279282
public static final String MERGE_OPERATION = "MERGE";
280283
public static final String OPTIMIZE_OPERATION = "OPTIMIZE";
@@ -1225,6 +1228,80 @@ public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle
12251228
}
12261229
}
12271230

1231+
@Override
1232+
public void renameColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle, String newColumnName)
1233+
{
1234+
DeltaLakeTableHandle table = (DeltaLakeTableHandle) tableHandle;
1235+
DeltaLakeColumnHandle deltaLakeColumn = (DeltaLakeColumnHandle) columnHandle;
1236+
String sourceColumnName = deltaLakeColumn.getName();
1237+
1238+
ColumnMappingMode columnMappingMode = getColumnMappingMode(table.getMetadataEntry());
1239+
if (columnMappingMode != ColumnMappingMode.NAME && columnMappingMode != ColumnMappingMode.ID) {
1240+
throw new TrinoException(NOT_SUPPORTED, "Cannot rename column with the column mapping: " + columnMappingMode);
1241+
}
1242+
1243+
ConnectorTableMetadata tableMetadata = getTableMetadata(session, table);
1244+
long commitVersion = table.getReadVersion() + 1;
1245+
List<String> partitionColumns = getPartitionedBy(tableMetadata.getProperties()).stream()
1246+
.map(columnName -> columnName.equals(sourceColumnName) ? newColumnName : columnName)
1247+
.collect(toImmutableList());
1248+
1249+
List<DeltaLakeColumnHandle> columns = tableMetadata.getColumns().stream()
1250+
.filter(column -> !column.isHidden())
1251+
.map(column -> toColumnHandle(
1252+
column.getName().equals(sourceColumnName) ? ColumnMetadata.builderFrom(column).setName(newColumnName).build() : column,
1253+
column.getName().equals(sourceColumnName) ? newColumnName : column.getName(),
1254+
column.getType(),
1255+
partitionColumns))
1256+
.collect(toImmutableList());
1257+
Map<String, String> columnComments = getColumnComments(table.getMetadataEntry()).entrySet().stream()
1258+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1259+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1260+
Map<String, Boolean> columnsNullability = getColumnsNullability(table.getMetadataEntry()).entrySet().stream()
1261+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1262+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1263+
Map<String, Map<String, Object>> columnMetadata = getColumnsMetadata(table.getMetadataEntry()).entrySet().stream()
1264+
.map(column -> column.getKey().equals(sourceColumnName) ? Map.entry(newColumnName, column.getValue()) : column)
1265+
.collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
1266+
try {
1267+
TransactionLogWriter transactionLogWriter = transactionLogWriterFactory.newWriter(session, table.getLocation());
1268+
appendTableEntries(
1269+
commitVersion,
1270+
transactionLogWriter,
1271+
table.getMetadataEntry().getId(),
1272+
columns,
1273+
partitionColumns,
1274+
columnComments,
1275+
columnsNullability,
1276+
columnMetadata,
1277+
table.getMetadataEntry().getConfiguration(),
1278+
RENAME_COLUMN_OPERATION,
1279+
session,
1280+
nodeVersion,
1281+
nodeId,
1282+
Optional.ofNullable(table.getMetadataEntry().getDescription()),
1283+
getProtocolEntry(session, table.getSchemaTableName()));
1284+
transactionLogWriter.flush();
1285+
1286+
statisticsAccess.readExtendedStatistics(session, table.getLocation()).ifPresent(existingStatistics -> {
1287+
ExtendedStatistics statistics = new ExtendedStatistics(
1288+
existingStatistics.getModelVersion(),
1289+
existingStatistics.getAlreadyAnalyzedModifiedTimeMax(),
1290+
existingStatistics.getColumnStatistics().entrySet().stream()
1291+
.map(stats -> stats.getKey().equals(sourceColumnName)
1292+
? Map.entry(newColumnName, DeltaLakeColumnStatistics.create(stats.getValue().getTotalSizeInBytes(), stats.getValue().getNdvSummary()))
1293+
: stats)
1294+
.collect(toImmutableMap(Entry::getKey, Entry::getValue)),
1295+
existingStatistics.getAnalyzedColumns()
1296+
.map(analyzedColumns -> analyzedColumns.stream().map(column -> column.equals(sourceColumnName) ? newColumnName : column).collect(toImmutableSet())));
1297+
statisticsAccess.updateExtendedStatistics(session, table.getLocation(), statistics);
1298+
});
1299+
}
1300+
catch (Exception e) {
1301+
throw new TrinoException(DELTA_LAKE_BAD_WRITE, format("Unable to rename '%s' column for: %s.%s", sourceColumnName, table.getSchemaName(), table.getTableName()), e);
1302+
}
1303+
}
1304+
12281305
private static void appendTableEntries(
12291306
long commitVersion,
12301307
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
@@ -112,7 +112,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
112112
return false;
113113

114114
case SUPPORTS_DROP_COLUMN:
115-
case SUPPORTS_RENAME_COLUMN:
116115
case SUPPORTS_SET_COLUMN_TYPE:
117116
return false;
118117

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

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

‎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
@@ -84,6 +84,7 @@ public final class TestGroups
8484
public static final String DELTA_LAKE_DATABRICKS = "delta-lake-databricks";
8585
public static final String DELTA_LAKE_EXCLUDE_73 = "delta-lake-exclude-73";
8686
public static final String DELTA_LAKE_EXCLUDE_91 = "delta-lake-exclude-91";
87+
public static final String DELTA_LAKE_EXCLUDE_104 = "delta-lake-exclude-104";
8788
public static final String DELTA_LAKE_EXCLUDE_113 = "delta-lake-exclude-113";
8889
public static final String HUDI = "hudi";
8990
public static final String PARQUET = "parquet";

0 commit comments

Comments
 (0)