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

Delta Lake checkpoint cleanups #21195

Merged
merged 8 commits into from
Mar 22, 2024
+178 −822
Merged

Delta Lake checkpoint cleanups #21195

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
11b395e
Remove unnecessary usage of BigInteger for LastCheckpoint
electrum Mar 22, 2024
cfd014d
Convert LastCheckpoint to record
electrum Mar 22, 2024
fb647c3
Convert CheckpointEntries to record
electrum Mar 22, 2024
39f7710
Convert ProtocolEntry to record
electrum Mar 22, 2024
1e26a62
Convert TransactionEntry to record
electrum Mar 22, 2024
d83e781
Convert CommitInfoEntry to record
electrum Mar 22, 2024
0a5f74c
Convert MetadataEntry.Format to record
electrum Mar 22, 2024
75d9f50
Convert RemoveFileEntry to record
electrum Mar 22, 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
20 changes: 10 additions & 10 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeHistoryTable.java
View file
Original file line number Diff line number Diff line change
@@ -221,20 +221,20 @@ private List<Page> buildPages(ConnectorSession session, List<CommitInfoEntry> co
commitInfoEntries.forEach(commitInfoEntry -> {
pagesBuilder.beginRow();

pagesBuilder.appendBigint(commitInfoEntry.getVersion());
pagesBuilder.appendTimestampTzMillis(commitInfoEntry.getTimestamp(), timeZoneKey);
write(commitInfoEntry.getUserId(), pagesBuilder);
write(commitInfoEntry.getUserName(), pagesBuilder);
write(commitInfoEntry.getOperation(), pagesBuilder);
if (commitInfoEntry.getOperationParameters() == null) {
pagesBuilder.appendBigint(commitInfoEntry.version());
pagesBuilder.appendTimestampTzMillis(commitInfoEntry.timestamp(), timeZoneKey);
write(commitInfoEntry.userId(), pagesBuilder);
write(commitInfoEntry.userName(), pagesBuilder);
write(commitInfoEntry.operation(), pagesBuilder);
if (commitInfoEntry.operationParameters() == null) {
pagesBuilder.appendNull();
}
else {
pagesBuilder.appendVarcharVarcharMap(commitInfoEntry.getOperationParameters());
pagesBuilder.appendVarcharVarcharMap(commitInfoEntry.operationParameters());
}
write(commitInfoEntry.getClusterId(), pagesBuilder);
pagesBuilder.appendBigint(commitInfoEntry.getReadVersion());
write(commitInfoEntry.getIsolationLevel(), pagesBuilder);
write(commitInfoEntry.clusterId(), pagesBuilder);
pagesBuilder.appendBigint(commitInfoEntry.readVersion());
write(commitInfoEntry.isolationLevel(), pagesBuilder);
commitInfoEntry.isBlindAppend().ifPresentOrElse(pagesBuilder::appendBoolean, pagesBuilder::appendNull);

pagesBuilder.endRow();
24 changes: 12 additions & 12 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
View file
Original file line number Diff line number Diff line change
@@ -569,11 +569,11 @@ public LocatedTableHandle getTableHandle(ConnectorSession session, SchemaTableNa
if (protocolEntry == null) {
return new CorruptedDeltaLakeTableHandle(tableName, managed, tableLocation, new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Protocol not found in transaction log for " + tableSnapshot.getTable()));
}
if (protocolEntry.getMinReaderVersion() > MAX_READER_VERSION) {
LOG.debug("Skip %s because the reader version is unsupported: %d", tableName, protocolEntry.getMinReaderVersion());
if (protocolEntry.minReaderVersion() > MAX_READER_VERSION) {
LOG.debug("Skip %s because the reader version is unsupported: %d", tableName, protocolEntry.minReaderVersion());
return null;
}
Set<String> unsupportedReaderFeatures = unsupportedReaderFeatures(protocolEntry.getReaderFeatures().orElse(ImmutableSet.of()));
Set<String> unsupportedReaderFeatures = unsupportedReaderFeatures(protocolEntry.readerFeatures().orElse(ImmutableSet.of()));
if (!unsupportedReaderFeatures.isEmpty()) {
LOG.debug("Skip %s because the table contains unsupported reader features: %s", tableName, unsupportedReaderFeatures);
return null;
@@ -1461,14 +1461,14 @@ private ProtocolEntry buildProtocolEntryForNewColumn(ProtocolEntry protocolEntry
}

return new ProtocolEntry(
max(protocolEntry.getMinReaderVersion(), TIMESTAMP_NTZ_SUPPORTED_READER_VERSION),
max(protocolEntry.getMinWriterVersion(), TIMESTAMP_NTZ_SUPPORTED_WRITER_VERSION),
max(protocolEntry.minReaderVersion(), TIMESTAMP_NTZ_SUPPORTED_READER_VERSION),
max(protocolEntry.minWriterVersion(), TIMESTAMP_NTZ_SUPPORTED_WRITER_VERSION),
Optional.of(ImmutableSet.<String>builder()
.addAll(protocolEntry.getReaderFeatures().orElse(ImmutableSet.of()))
.addAll(protocolEntry.readerFeatures().orElse(ImmutableSet.of()))
.add(TIMESTAMP_NTZ_FEATURE_NAME)
.build()),
Optional.of(ImmutableSet.<String>builder()
.addAll(protocolEntry.getWriterFeatures().orElse(ImmutableSet.of()))
.addAll(protocolEntry.writerFeatures().orElse(ImmutableSet.of()))
.add(TIMESTAMP_NTZ_FEATURE_NAME)
.build()));
}
@@ -2341,7 +2341,7 @@ private void checkWriteSupported(DeltaLakeTableHandle handle)

private static void checkUnsupportedWriterFeatures(ProtocolEntry protocolEntry)
{
Set<String> unsupportedWriterFeatures = unsupportedWriterFeatures(protocolEntry.getWriterFeatures().orElse(ImmutableSet.of()));
Set<String> unsupportedWriterFeatures = unsupportedWriterFeatures(protocolEntry.writerFeatures().orElse(ImmutableSet.of()));
if (!unsupportedWriterFeatures.isEmpty()) {
throw new TrinoException(NOT_SUPPORTED, "Unsupported writer features: " + unsupportedWriterFeatures);
}
@@ -2357,7 +2357,7 @@ private void checkUnsupportedGeneratedColumns(MetadataEntry metadataEntry)

private void checkSupportedWriterVersion(DeltaLakeTableHandle handle)
{
int requiredWriterVersion = handle.getProtocolEntry().getMinWriterVersion();
int requiredWriterVersion = handle.getProtocolEntry().minWriterVersion();
if (requiredWriterVersion > MAX_WRITER_VERSION) {
throw new TrinoException(
NOT_SUPPORTED,
@@ -2528,7 +2528,7 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta

long createdTime = Instant.now().toEpochMilli();

int requiredWriterVersion = currentProtocolEntry.getMinWriterVersion();
int requiredWriterVersion = currentProtocolEntry.minWriterVersion();
Optional<MetadataEntry> metadataEntry = Optional.empty();
if (properties.containsKey(CHANGE_DATA_FEED_ENABLED_PROPERTY)) {
boolean changeDataFeedEnabled = (Boolean) properties.get(CHANGE_DATA_FEED_ENABLED_PROPERTY)
@@ -2550,8 +2550,8 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta
long commitVersion = readVersion + 1;

Optional<ProtocolEntry> protocolEntry = Optional.empty();
if (requiredWriterVersion != currentProtocolEntry.getMinWriterVersion()) {
protocolEntry = Optional.of(new ProtocolEntry(currentProtocolEntry.getMinReaderVersion(), requiredWriterVersion, currentProtocolEntry.getReaderFeatures(), currentProtocolEntry.getWriterFeatures()));
if (requiredWriterVersion != currentProtocolEntry.minWriterVersion()) {
protocolEntry = Optional.of(new ProtocolEntry(currentProtocolEntry.minReaderVersion(), requiredWriterVersion, currentProtocolEntry.readerFeatures(), currentProtocolEntry.writerFeatures()));
}

try {
8 changes: 4 additions & 4 deletions ...in/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakePropertiesTable.java
View file
Original file line number Diff line number Diff line change
@@ -107,17 +107,17 @@ private List<Page> buildPages(MetadataEntry metadataEntry, ProtocolEntry protoco

pagesBuilder.beginRow();
pagesBuilder.appendVarchar(MIN_READER_VERSION_KEY);
pagesBuilder.appendVarchar(String.valueOf(protocolEntry.getMinReaderVersion()));
pagesBuilder.appendVarchar(String.valueOf(protocolEntry.minReaderVersion()));
pagesBuilder.endRow();

pagesBuilder.beginRow();
pagesBuilder.appendVarchar(MIN_WRITER_VERSION_KEY);
pagesBuilder.appendVarchar(String.valueOf(protocolEntry.getMinWriterVersion()));
pagesBuilder.appendVarchar(String.valueOf(protocolEntry.minWriterVersion()));
pagesBuilder.endRow();

ImmutableSet.<String>builder()
.addAll(protocolEntry.getReaderFeatures().orElseGet(ImmutableSet::of))
.addAll(protocolEntry.getWriterFeatures().orElseGet(ImmutableSet::of))
.addAll(protocolEntry.readerFeatures().orElseGet(ImmutableSet::of))
.addAll(protocolEntry.writerFeatures().orElseGet(ImmutableSet::of))
.build().forEach(feature -> {
pagesBuilder.beginRow();
pagesBuilder.appendVarchar(DELTA_FEATURE_PREFIX + feature);
6 changes: 3 additions & 3 deletions ...c/main/java/io/trino/plugin/deltalake/functions/tablechanges/TableChangesSplitSource.java
View file
Original file line number Diff line number Diff line change
@@ -102,7 +102,7 @@ private Stream<ConnectorSplit> prepareSplits(long currentVersion, long tableRead
cdcEntry.getPath(),
cdcEntry.getCanonicalPartitionValues()));
}
if (entry.getRemove() != null && entry.getRemove().isDataChange()) {
if (entry.getRemove() != null && entry.getRemove().dataChange()) {
containsRemoveEntry = true;
}
}
@@ -154,9 +154,9 @@ private TableChangesSplit mapToDeltaLakeTableChangesSplit(
path,
length,
canonicalPartitionValues,
commitInfoEntry.getTimestamp(),
commitInfoEntry.timestamp(),
source,
commitInfoEntry.getVersion());
commitInfoEntry.version());
}

@Override
8 changes: 4 additions & 4 deletions ...n/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/procedure/VacuumProcedure.java
View file
Original file line number Diff line number Diff line change
@@ -183,10 +183,10 @@ private void doVacuum(

TableSnapshot tableSnapshot = metadata.getSnapshot(session, tableName, handle.getLocation(), handle.getReadVersion());
ProtocolEntry protocolEntry = transactionLogAccess.getProtocolEntry(session, tableSnapshot);
if (protocolEntry.getMinWriterVersion() > MAX_WRITER_VERSION) {
throw new TrinoException(NOT_SUPPORTED, "Cannot execute vacuum procedure with %d writer version".formatted(protocolEntry.getMinWriterVersion()));
if (protocolEntry.minWriterVersion() > MAX_WRITER_VERSION) {
throw new TrinoException(NOT_SUPPORTED, "Cannot execute vacuum procedure with %d writer version".formatted(protocolEntry.minWriterVersion()));
}
Set<String> unsupportedWriterFeatures = unsupportedWriterFeatures(protocolEntry.getWriterFeatures().orElse(ImmutableSet.of()));
Set<String> unsupportedWriterFeatures = unsupportedWriterFeatures(protocolEntry.writerFeatures().orElse(ImmutableSet.of()));
if (!unsupportedWriterFeatures.isEmpty()) {
throw new TrinoException(NOT_SUPPORTED, "Cannot execute vacuum procedure with %s writer features".formatted(unsupportedWriterFeatures));
}
@@ -221,7 +221,7 @@ private void doVacuum(
.collect(toImmutableList()))
.map(DeltaLakeTransactionLogEntry::getRemove)
.filter(Objects::nonNull)
.map(RemoveFileEntry::getPath))
.map(RemoveFileEntry::path))
.peek(path -> checkState(!path.startsWith(tableLocation), "Unexpected absolute path in transaction log: %s", path))
.collect(toImmutableSet());
}
Loading