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

Derive name mapping from table schema when iceberg.id is missing #24795

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from
+14 −12
Draft

Derive name mapping from table schema when iceberg.id is missing #24795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
18 changes: 10 additions & 8 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.mapping.MappedField;
import org.apache.iceberg.mapping.MappedFields;
import org.apache.iceberg.mapping.MappingUtil;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
Expand Down Expand Up @@ -265,7 +266,7 @@ public ConnectorPageSource createPageSource(
split.getFileFormat(),
split.getFileIoProperties(),
split.getDataSequenceNumber(),
tableHandle.getNameMappingJson().map(NameMappingParser::fromJson));
tableHandle.getNameMappingJson().map(NameMappingParser::fromJson).orElseGet(() -> MappingUtil.create(schema)));
}

public ConnectorPageSource createPageSource(
Expand All @@ -287,7 +288,7 @@ public ConnectorPageSource createPageSource(
IcebergFileFormat fileFormat,
Map<String, String> fileIoProperties,
long dataSequenceNumber,
Optional<NameMapping> nameMapping)
NameMapping nameMapping)
{
Set<IcebergColumnHandle> deleteFilterRequiredColumns = requiredColumnsForDeletes(tableSchema, deletes);
Map<Integer, Optional<String>> partitionKeys = getPartitionKeys(partitionData, partitionSpec);
Expand Down Expand Up @@ -465,17 +466,18 @@ private ConnectorPageSource openDeletes(
List<IcebergColumnHandle> columns,
TupleDomain<IcebergColumnHandle> tupleDomain)
{
Schema schema = schemaFromHandles(columns);
return createDataPageSource(
session,
fileSystem.newInputFile(Location.of(delete.path()), delete.fileSizeInBytes()),
0,
delete.fileSizeInBytes(),
delete.fileSizeInBytes(),
IcebergFileFormat.fromIceberg(delete.format()),
schemaFromHandles(columns),
schema,
columns,
tupleDomain,
Optional.empty(),
MappingUtil.create(schema),
ImmutableMap.of())
.readerPageSource()
.get();
Expand All @@ -491,7 +493,7 @@ private ReaderPageSourceWithRowPositions createDataPageSource(
Schema fileSchema,
List<IcebergColumnHandle> dataColumns,
TupleDomain<IcebergColumnHandle> predicate,
Optional<NameMapping> nameMapping,
NameMapping nameMapping,
Map<Integer, Optional<String>> partitionKeys)
{
return switch (fileFormat) {
Expand Down Expand Up @@ -531,14 +533,14 @@ private ReaderPageSourceWithRowPositions createDataPageSource(
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session)),
predicate,
fileFormatDataSourceStats,
nameMapping,
Optional.of(nameMapping),
partitionKeys);
case AVRO -> createAvroPageSource(
inputFile,
start,
length,
fileSchema,
nameMapping,
Optional.of(nameMapping),
dataColumns);
};
}
Expand Down Expand Up @@ -587,7 +589,7 @@ private static ReaderPageSourceWithRowPositions createOrcPageSource(
OrcReaderOptions options,
FileFormatDataSourceStats stats,
TypeManager typeManager,
Optional<NameMapping> nameMapping,
NameMapping nameMapping,
Map<Integer, Optional<String>> partitionKeys)
{
OrcDataSource orcDataSource = null;
Expand Down
6 changes: 3 additions & 3 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/util/OrcIcebergIds.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,13 @@ public final class OrcIcebergIds
{
private OrcIcebergIds() {}

public static Map<Integer, OrcColumn> fileColumnsByIcebergId(OrcReader reader, Optional<NameMapping> nameMapping)
public static Map<Integer, OrcColumn> fileColumnsByIcebergId(OrcReader reader, NameMapping nameMapping)
{
List<OrcColumn> fileColumns = reader.getRootColumn().getNestedColumns();

if (nameMapping.isPresent() && !hasIds(reader.getRootColumn())) {
if (!hasIds(reader.getRootColumn())) {
fileColumns = fileColumns.stream()
.map(orcColumn -> setMissingFieldIds(orcColumn, nameMapping.get(), ImmutableList.of(orcColumn.getColumnName())))
.map(orcColumn -> setMissingFieldIds(orcColumn, nameMapping, ImmutableList.of(orcColumn.getColumnName())))
.collect(toImmutableList());
}

Expand Down
2 changes: 1 addition & 1 deletion plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/util/OrcMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ public static Metrics fileMetrics(TrinoInputFile file, MetricsConfig metricsConf
Footer footer = reader.get().getFooter();

// use name mapping to compute missing Iceberg field IDs
Optional<NameMapping> nameMapping = Optional.of(MappingUtil.create(schema));
NameMapping nameMapping = MappingUtil.create(schema);
Copy link
Member

Choose a reason for hiding this comment

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

So if I'm understanding correctly, we have two different codepaths that eventually land at fileColumnsByIcebergId. One from IcebergPageSourceProvider, and one here in OrcMetrics. Prior to this PR, the latter would use MappingUtil to derive the schema, but the former would only use the table properties. And in this PR, we are standardizing so that both codepaths use MappingUtil. Do I have that right?

I'm curious, doesn't this OrcMetrics codepath also need to handle the case where the name mapping is set as a JSON in the table properties? Like why aren't these codepaths completely unified in how they derive a NameMapping from the combination of ORC files / table metadata / table schema?

Copy link
Member Author

@weijiii weijiii Jan 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

IIUC OrcMetrics::fileMetrics is only used when migrating table to Iceberg, therefore the Iceberg table metadata is not yet created as well as the table properties. Column Ids are expected to be missing in the ORC files so we are using the name mapping derived from table schema to fill the gap.

Copy link
Member

Choose a reason for hiding this comment

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

Makes sense, thanks for clarifying!

Map<OrcColumnId, OrcColumn> mappedColumns = fileColumnsByIcebergId(reader.get(), nameMapping)
.values().stream()
.collect(toImmutableMap(OrcColumn::getColumnId, identity()));
Expand Down
Loading