Skip to content

Support inserting into Delta Lake table having invariants #16136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ebyhr merged 2 commits into from
Mar 9, 2023
Merged

Support inserting into Delta Lake table having invariants #16136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
84b9bc0
Rename variable in DeltaLakeSchemaSupport
ebyhr Feb 16, 2023
c22bab7
Support inserting into Delta Lake table having invariants
ebyhr Feb 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 7 additions & 8 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -471,9 +471,10 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
String location = metastore.getTableLocation(tableHandle.getSchemaTableName(), session);
Map<String, String> columnComments = getColumnComments(tableHandle.getMetadataEntry());
Map<String, Boolean> columnsNullability = getColumnsNullability(tableHandle.getMetadataEntry());
List<String> checkConstraints = getCheckConstraints(tableHandle.getMetadataEntry()).values().stream()
.map(SparkExpressionParser::toTrinoExpression)
.collect(toImmutableList());
List<String> constraints = ImmutableList.<String>builder()
.addAll(getCheckConstraints(tableHandle.getMetadataEntry()).values())
.addAll(getColumnInvariants(tableHandle.getMetadataEntry()).values()) // The internal logic for column invariants in Delta Lake is same as check constraints
.build();
List<ColumnMetadata> columns = getColumns(tableHandle.getMetadataEntry()).stream()
.map(column -> getColumnMetadata(column, columnComments.get(column.getName()), columnsNullability.getOrDefault(column.getName(), true)))
.collect(toImmutableList());
Expand All @@ -497,7 +498,9 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
columns,
properties.buildOrThrow(),
Optional.ofNullable(tableHandle.getMetadataEntry().getDescription()),
checkConstraints);
constraints.stream()
.map(SparkExpressionParser::toTrinoExpression)
.collect(toImmutableList()));
}

@Override
Expand Down Expand Up @@ -1270,10 +1273,6 @@ public ConnectorInsertTableHandle beginInsert(ConnectorSession session, Connecto
format("Inserts are not enabled on the %1$s filesystem in order to avoid eventual data corruption which may be caused by concurrent data modifications on the table. " +
"Writes to the %1$s filesystem can be however enabled with the '%2$s' configuration property.", fileSystem, ENABLE_NON_CONCURRENT_WRITES_CONFIGURATION_KEY));
}
Map<String, String> columnInvariants = getColumnInvariants(table.getMetadataEntry());
if (!columnInvariants.isEmpty()) {
throw new TrinoException(NOT_SUPPORTED, "Inserts are not supported for tables with delta invariants");
}
checkUnsupportedGeneratedColumns(table.getMetadataEntry());
checkSupportedWriterVersion(session, table.getSchemaTableName());

Expand Down
16 changes: 13 additions & 3 deletions ...a-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/DeltaLakeSchemaSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -410,7 +410,17 @@ public static Map<String, String> getColumnInvariants(MetadataEntry metadataEntr
private static String getInvariants(JsonNode node)
{
JsonNode invariants = node.get("metadata").get("delta.invariants");
return invariants == null ? null : invariants.asText();
return invariants == null ? null : extractInvariantsExpression(invariants.asText());
}

private static String extractInvariantsExpression(String invariants)
{
try {
return OBJECT_MAPPER.readTree(invariants).get("expression").get("expression").asText();
}
catch (JsonProcessingException e) {
throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, getLocation(e), "Failed to parse invariants expression: " + invariants, e);
}
}

public static Map<String, String> getGeneratedColumnExpressions(MetadataEntry metadataEntry)
Expand All @@ -421,8 +431,8 @@ public static Map<String, String> getGeneratedColumnExpressions(MetadataEntry me
@Nullable
private static String getGeneratedColumnExpressions(JsonNode node)
{
JsonNode invariants = node.get("metadata").get("delta.generationExpression");
return invariants == null ? null : invariants.asText();
JsonNode generationExpression = node.get("metadata").get("delta.generationExpression");
return generationExpression == null ? null : generationExpression.asText();
}

public static Map<String, String> getCheckConstraints(MetadataEntry metadataEntry)
Expand Down
32 changes: 22 additions & 10 deletions ...o-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorSmokeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,13 +160,22 @@ public void testWritesLockInvalidContents(String writeStatement, String expected
}

@Test
public void testReadingTableWithDeltaColumnInvariant()
public void testDeltaColumnInvariant()
{
assertThat(getQueryRunner().execute("SELECT * FROM invariants").getRowCount()).isEqualTo(1);
assertThatThrownBy(() -> query("INSERT INTO invariants VALUES(2)"))
.hasMessageContaining("Inserts are not supported for tables with delta invariants");
assertThatThrownBy(() -> query("UPDATE invariants SET dummy = 3 WHERE dummy = 1"))
.hasMessageContaining("Updates are not supported for tables with delta invariants");
String tableName = "test_invariants_" + randomNameSuffix();
hiveMinioDataLake.copyResources("databricks/invariants", tableName);
assertUpdate("CALL system.register_table('%s', '%s', '%s')".formatted(SCHEMA, tableName, getLocationForTable(bucketName, tableName)));

assertQuery("SELECT * FROM " + tableName, "VALUES 1");
assertUpdate("INSERT INTO " + tableName + " VALUES(2)", 1);
Comment thread
ebyhr marked this conversation as resolved.
Outdated
assertQuery("SELECT * FROM " + tableName, "VALUES (1), (2)");

assertThatThrownBy(() -> query("INSERT INTO " + tableName + " VALUES(3)"))
.hasMessageContaining("Check constraint violation: (\"dummy\" < 3)");
assertThatThrownBy(() -> query("UPDATE " + tableName + " SET dummy = 3 WHERE dummy = 1"))
.hasMessageContaining("Updating a table with a check constraint is not supported");

assertQuery("SELECT * FROM " + tableName, "VALUES (1), (2)");
}

@Test
Expand All @@ -180,15 +189,18 @@ public void testSchemaEvolutionOnTableWithColumnInvariant()
tableName,
getLocationForTable(bucketName, tableName)));

assertThatThrownBy(() -> query("INSERT INTO invariants VALUES(2)"))
.hasMessageContaining("Inserts are not supported for tables with delta invariants");
Comment on lines 183 to 184
Copy link
Copy Markdown
Member

@findepi findepi Mar 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How was that working? Shouldn't it be failing with table not found?

Copy link
Copy Markdown
Member Author

@ebyhr ebyhr Mar 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if I understand the question correctly. The previous table name was invariants having column invariants and it existed, so it shouldn't throw table not found in my opinion. I renamed to a new table name with random suffix to avoid modifying shared tables.

assertThatThrownBy(() -> query("INSERT INTO " + tableName + " VALUES(3)"))
.hasMessageContaining("Check constraint violation: (\"dummy\" < 3)");

assertUpdate("ALTER TABLE " + tableName + " ADD COLUMN c INT");
assertUpdate("COMMENT ON COLUMN " + tableName + ".c IS 'example column comment'");
assertUpdate("COMMENT ON TABLE " + tableName + " IS 'example table comment'");

assertThatThrownBy(() -> query("INSERT INTO " + tableName + " VALUES(2, 2)"))
Comment thread
ebyhr marked this conversation as resolved.
Outdated
.hasMessageContaining("Inserts are not supported for tables with delta invariants");
assertThatThrownBy(() -> query("INSERT INTO " + tableName + " VALUES(3, 30)"))
.hasMessageContaining("Check constraint violation: (\"dummy\" < 3)");

assertUpdate("INSERT INTO " + tableName + " VALUES(2, 20)", 1);
assertQuery("SELECT * FROM " + tableName, "VALUES (1, NULL), (2, 20)");
}

@DataProvider
Expand Down