Skip to content
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

Minor refactoring in Delta Lake connector #20955

Merged
merged 2 commits into from
Mar 8, 2024
Merged

Minor refactoring in Delta Lake connector #20955

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
d636f20
Remove unused code from Delta Lake
ebyhr Mar 6, 2024
5cbfd0a
Move ConnectorSession to the 1st argument in TransactionLogAccess
ebyhr Mar 6, 2024
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
2 changes: 1 addition & 1 deletion plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeHistoryTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ public ConnectorPageSource pageSource(ConnectorTransactionHandle transactionHand
SchemaTableName baseTableName = new SchemaTableName(tableName.getSchemaName(), DeltaLakeTableName.tableNameFrom(tableName.getTableName()));
TableSnapshot tableSnapshot = transactionLogAccess.loadSnapshot(session, baseTableName, tableLocation);
snapshotVersion = tableSnapshot.getVersion();
transactionLogAccess.getMetadataEntry(tableSnapshot, session);
transactionLogAccess.getMetadataEntry(session, tableSnapshot);
}
catch (IOException e) {
throw new TrinoException(DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA, "Unable to load table metadata from location: " + tableLocation, e);
Expand Down
23 changes: 9 additions & 14 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 @@ -771,7 +771,7 @@ public Iterator<TableColumnsMetadata> streamTableColumns(ConnectorSession sessio
}
String tableLocation = metastoreTable.get().location();
TableSnapshot snapshot = transactionLogAccess.loadSnapshot(session, table, tableLocation);
MetadataEntry metadata = transactionLogAccess.getMetadataEntry(snapshot, session);
MetadataEntry metadata = transactionLogAccess.getMetadataEntry(session, snapshot);
ProtocolEntry protocol = transactionLogAccess.getProtocolEntry(session, snapshot);
List<ColumnMetadata> columnMetadata = getTableColumnMetadata(metadata, protocol);
return Stream.of(TableColumnsMetadata.forTable(table, columnMetadata));
Expand Down Expand Up @@ -966,7 +966,7 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
// As a precaution, clear the caches
statisticsAccess.invalidateCache(schemaTableName, Optional.of(location));
transactionLogAccess.invalidateCache(schemaTableName, Optional.of(location));
metastore.createTable(session, table, principalPrivileges);
metastore.createTable(table, principalPrivileges);
}

public static Table buildTable(ConnectorSession session, SchemaTableName schemaTableName, String location, boolean isExternal)
Expand Down Expand Up @@ -1261,7 +1261,7 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(
// As a precaution, clear the caches
statisticsAccess.invalidateCache(schemaTableName, Optional.of(location));
transactionLogAccess.invalidateCache(schemaTableName, Optional.of(location));
metastore.createTable(session, table, principalPrivileges);
metastore.createTable(table, principalPrivileges);
}
catch (Exception e) {
// Remove the transaction log entry if the table creation fails
Expand Down Expand Up @@ -1379,7 +1379,7 @@ public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle

if (!newColumnMetadata.isNullable()) {
boolean tableHasDataFiles;
try (Stream<AddFileEntry> addFileEntries = transactionLogAccess.getActiveFiles(getSnapshot(session, handle), handle.getMetadataEntry(), handle.getProtocolEntry(), session)) {
try (Stream<AddFileEntry> addFileEntries = transactionLogAccess.getActiveFiles(session, getSnapshot(session, handle), handle.getMetadataEntry(), handle.getProtocolEntry())) {
tableHasDataFiles = addFileEntries.findAny().isPresent();
}
if (tableHasDataFiles) {
Expand Down Expand Up @@ -2427,7 +2427,7 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle
{
LocatedTableHandle handle = (LocatedTableHandle) tableHandle;
boolean deleteData = handle.managed();
metastore.dropTable(session, handle.schemaTableName(), handle.location(), deleteData);
metastore.dropTable(handle.schemaTableName(), handle.location(), deleteData);
if (deleteData) {
try {
fileSystemFactory.create(session).deleteDirectory(Location.of(handle.location()));
Expand All @@ -2450,7 +2450,7 @@ public void renameTable(ConnectorSession session, ConnectorTableHandle tableHand
if (table.managed() && !allowManagedTableRename) {
throw new TrinoException(NOT_SUPPORTED, "Renaming managed tables is not allowed with current metastore configuration");
}
metastore.renameTable(session, handle.getSchemaTableName(), newTableName);
metastore.renameTable(handle.getSchemaTableName(), newTableName);
}

private CommitInfoEntry getCommitInfoEntry(
Expand Down Expand Up @@ -3200,7 +3200,7 @@ private void generateMissingFileStatistics(ConnectorSession session, DeltaLakeTa
{
Map<String, AddFileEntry> addFileEntriesWithNoStats;
try (Stream<AddFileEntry> activeFiles = transactionLogAccess.getActiveFiles(
getSnapshot(session, tableHandle), tableHandle.getMetadataEntry(), tableHandle.getProtocolEntry(), session)) {
session, getSnapshot(session, tableHandle), tableHandle.getMetadataEntry(), tableHandle.getProtocolEntry())) {
addFileEntriesWithNoStats = activeFiles.filter(addFileEntry -> addFileEntry.getStats().isEmpty()
|| addFileEntry.getStats().get().getNumRecords().isEmpty()
|| addFileEntry.getStats().get().getMaxValues().isEmpty()
Expand Down Expand Up @@ -3563,12 +3563,12 @@ private Stream<AddFileEntry> getAddFileEntriesMatchingEnforcedPartitionConstrain
{
TableSnapshot tableSnapshot = getSnapshot(session, tableHandle);
Stream<AddFileEntry> validDataFiles = transactionLogAccess.getActiveFiles(
session,
tableSnapshot,
tableHandle.getMetadataEntry(),
tableHandle.getProtocolEntry(),
tableHandle.getEnforcedPartitionConstraint(),
tableHandle.getProjectedColumns(),
session);
tableHandle.getProjectedColumns());
TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint = tableHandle.getEnforcedPartitionConstraint();
if (enforcedPartitionConstraint.isAll()) {
return validDataFiles;
Expand Down Expand Up @@ -3808,9 +3808,4 @@ private static Optional<String> getQueryId(Database database)
{
return Optional.ofNullable(database.getParameters().get(TRINO_QUERY_ID_NAME));
}

public static Optional<String> getQueryId(Table table)
{
return Optional.ofNullable(table.getParameters().get(TRINO_QUERY_ID_NAME));
}
}
2 changes: 1 addition & 1 deletion ...in/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakePropertiesTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ public ConnectorPageSource pageSource(ConnectorTransactionHandle transactionHand
try {
SchemaTableName baseTableName = new SchemaTableName(tableName.getSchemaName(), DeltaLakeTableName.tableNameFrom(tableName.getTableName()));
TableSnapshot tableSnapshot = transactionLogAccess.loadSnapshot(session, baseTableName, tableLocation);
metadataEntry = transactionLogAccess.getMetadataEntry(tableSnapshot, session);
metadataEntry = transactionLogAccess.getMetadataEntry(session, tableSnapshot);
protocolEntry = transactionLogAccess.getProtocolEntry(session, tableSnapshot);
}
catch (IOException e) {
Expand Down
4 changes: 2 additions & 2 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSplitManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,12 +159,12 @@ private Stream<DeltaLakeSplit> getSplits(
TableSnapshot tableSnapshot = deltaLakeTransactionManager.get(transaction, session.getIdentity())
.getSnapshot(session, tableHandle.getSchemaTableName(), tableHandle.getLocation(), tableHandle.getReadVersion());
Stream<AddFileEntry> validDataFiles = transactionLogAccess.getActiveFiles(
session,
tableSnapshot,
tableHandle.getMetadataEntry(),
tableHandle.getProtocolEntry(),
tableHandle.getEnforcedPartitionConstraint(),
tableHandle.getProjectedColumns(),
session);
tableHandle.getProjectedColumns());
TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint = tableHandle.getEnforcedPartitionConstraint();
TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint = tableHandle.getNonPartitionConstraint();
Domain pathDomain = getPathDomain(nonPartitionConstraint);
Expand Down
7 changes: 3 additions & 4 deletions ...rino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/DeltaLakeMetastore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
import io.trino.plugin.hive.metastore.Database;
import io.trino.plugin.hive.metastore.PrincipalPrivileges;
import io.trino.plugin.hive.metastore.Table;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;

import java.util.List;
Expand All @@ -38,9 +37,9 @@ public interface DeltaLakeMetastore

void dropDatabase(String databaseName, boolean deleteData);

void createTable(ConnectorSession session, Table table, PrincipalPrivileges principalPrivileges);
void createTable(Table table, PrincipalPrivileges principalPrivileges);

void dropTable(ConnectorSession session, SchemaTableName schemaTableName, String tableLocation, boolean deleteData);
void dropTable(SchemaTableName schemaTableName, String tableLocation, boolean deleteData);

void renameTable(ConnectorSession session, SchemaTableName from, SchemaTableName to);
void renameTable(SchemaTableName from, SchemaTableName to);
}
7 changes: 3 additions & 4 deletions .../main/java/io/trino/plugin/deltalake/metastore/HiveMetastoreBackedDeltaLakeMetastore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
import io.trino.plugin.hive.metastore.PrincipalPrivileges;
import io.trino.plugin.hive.metastore.Table;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.SchemaTableName;

import java.util.List;
Expand Down Expand Up @@ -108,19 +107,19 @@ public void dropDatabase(String databaseName, boolean deleteData)
}

@Override
public void createTable(ConnectorSession session, Table table, PrincipalPrivileges principalPrivileges)
public void createTable(Table table, PrincipalPrivileges principalPrivileges)
{
delegate.createTable(table, principalPrivileges);
}

@Override
public void dropTable(ConnectorSession session, SchemaTableName schemaTableName, String tableLocation, boolean deleteData)
public void dropTable(SchemaTableName schemaTableName, String tableLocation, boolean deleteData)
{
delegate.dropTable(schemaTableName.getSchemaName(), schemaTableName.getTableName(), deleteData);
}

@Override
public void renameTable(ConnectorSession session, SchemaTableName from, SchemaTableName to)
public void renameTable(SchemaTableName from, SchemaTableName to)
{
delegate.renameTable(from.getSchemaName(), from.getTableName(), to.getSchemaName(), to.getTableName());
}
Expand Down
4 changes: 2 additions & 2 deletions ...-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/RegisterTableProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ private void doRegisterTable(
// Verify we're registering a location with a valid table
try {
TableSnapshot tableSnapshot = transactionLogAccess.loadSnapshot(session, table.getSchemaTableName(), tableLocation);
transactionLogAccess.getMetadataEntry(tableSnapshot, session); // verify metadata exists
transactionLogAccess.getMetadataEntry(session, tableSnapshot); // verify metadata exists
}
catch (TrinoException e) {
throw e;
Expand All @@ -175,7 +175,7 @@ private void doRegisterTable(
throw new TrinoException(DELTA_LAKE_INVALID_TABLE, "Failed to access table location: " + tableLocation, e);
}

metastore.createTable(session, table, principalPrivileges);
metastore.createTable(table, principalPrivileges);
}
}
}
2 changes: 1 addition & 1 deletion ...elta-lake/src/main/java/io/trino/plugin/deltalake/procedure/UnregisterTableProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ private void doUnregisterTable(ConnectorAccessControl accessControl, ConnectorSe
if (tableHandle == null) {
throw new TableNotFoundException(schemaTableName);
}
metadata.getMetastore().dropTable(session, schemaTableName, tableHandle.location(), false);
metadata.getMetastore().dropTable(schemaTableName, tableHandle.location(), false);
// As a precaution, clear the caches
statisticsAccess.invalidateCache(schemaTableName, Optional.of(tableHandle.location()));
transactionLogAccess.invalidateCache(schemaTableName, Optional.of(tableHandle.location()));
Expand Down
2 changes: 1 addition & 1 deletion ...n/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/VacuumProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ private void doVacuum(
List<Long> recentVersions = transactionLogAccess.getPastTableVersions(fileSystem, transactionLogDir, threshold, tableSnapshot.getVersion());
Set<String> retainedPaths;
try (Stream<AddFileEntry> activeAddEntries = transactionLogAccess.getActiveFiles(
tableSnapshot, handle.getMetadataEntry(), handle.getProtocolEntry(), session)) {
session, tableSnapshot, handle.getMetadataEntry(), handle.getProtocolEntry())) {
retainedPaths = Stream.concat(
activeAddEntries
.map(AddFileEntry::getPath),
Expand Down
4 changes: 2 additions & 2 deletions .../src/main/java/io/trino/plugin/deltalake/statistics/FileBasedTableStatisticsProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,12 +113,12 @@ public TableStatistics getTableStatistics(ConnectorSession session, DeltaLakeTab
.collect(toImmutableList());

try (Stream<AddFileEntry> addEntries = transactionLogAccess.getActiveFiles(
session,
tableSnapshot,
tableHandle.getMetadataEntry(),
tableHandle.getProtocolEntry(),
tableHandle.getEnforcedPartitionConstraint(),
tableHandle.getProjectedColumns(),
session)) {
tableHandle.getProjectedColumns())) {
Iterator<AddFileEntry> addEntryIterator = addEntries.iterator();
while (addEntryIterator.hasNext()) {
AddFileEntry addEntry = addEntryIterator.next();
Expand Down
Loading