Skip to content

Don't truncate min/max for iceberg delete files #12671

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
findepi merged 1 commit into from
Jun 8, 2022
Merged

Don't truncate min/max for iceberg delete files #12671

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: 11 additions & 7 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergFileWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import io.airlift.units.DataSize;
import io.trino.orc.OrcDataSink;
import io.trino.orc.OrcDataSource;
import io.trino.orc.OrcDataSourceId;
Expand Down Expand Up @@ -131,10 +132,10 @@ public IcebergFileWriter createDataFileWriter(
{
switch (fileFormat) {
case PARQUET:
// TODO use metricsConfig
return createParquetWriter(outputPath, icebergSchema, jobConf, session, hdfsContext);
// TODO use metricsConfig https://github.com/trinodb/trino/issues/9791
Comment thread
findepi marked this conversation as resolved.
return createParquetWriter(MetricsConfig.getDefault(), outputPath, icebergSchema, jobConf, session, hdfsContext);
Copy link
Copy Markdown
Member

@alexjo2144 alexjo2144 Jun 3, 2022

Choose a reason for hiding this comment

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

Is there a GH Issue to support metrics mode for Parquet? If not, can you make one

Either way, link it here

case ORC:
return createOrcWriter(metricsConfig, outputPath, icebergSchema, jobConf, session, storageProperties);
return createOrcWriter(metricsConfig, outputPath, icebergSchema, jobConf, session, storageProperties, getOrcStringStatisticsLimit(session));
default:
throw new TrinoException(NOT_SUPPORTED, "File format not supported: " + fileFormat);
}
Expand All @@ -150,15 +151,16 @@ public IcebergFileWriter createPositionDeleteWriter(
{
switch (fileFormat) {
case PARQUET:
return createParquetWriter(outputPath, POSITION_DELETE_SCHEMA, jobConf, session, hdfsContext);
return createParquetWriter(FULL_METRICS_CONFIG, outputPath, POSITION_DELETE_SCHEMA, jobConf, session, hdfsContext);
case ORC:
return createOrcWriter(FULL_METRICS_CONFIG, outputPath, POSITION_DELETE_SCHEMA, jobConf, session, storageProperties);
return createOrcWriter(FULL_METRICS_CONFIG, outputPath, POSITION_DELETE_SCHEMA, jobConf, session, storageProperties, DataSize.ofBytes(Integer.MAX_VALUE));
default:
throw new TrinoException(NOT_SUPPORTED, "File format not supported: " + fileFormat);
}
}

private IcebergFileWriter createParquetWriter(
MetricsConfig metricsConfig,
Path outputPath,
Schema icebergSchema,
JobConf jobConf,
Expand Down Expand Up @@ -187,6 +189,7 @@ private IcebergFileWriter createParquetWriter(
.build();

return new IcebergParquetFileWriter(
metricsConfig,
hdfsEnvironment.doAs(session.getIdentity(), () -> fileSystem.create(outputPath)),
rollbackAction,
fileColumnTypes,
Expand All @@ -211,7 +214,8 @@ private IcebergFileWriter createOrcWriter(
Schema icebergSchema,
JobConf jobConf,
ConnectorSession session,
Map<String, String> storageProperties)
Map<String, String> storageProperties,
DataSize stringStatisticsLimit)
{
try {
FileSystem fileSystem = hdfsEnvironment.getFileSystem(session.getIdentity(), outputPath, jobConf);
Expand Down Expand Up @@ -261,7 +265,7 @@ private IcebergFileWriter createOrcWriter(
.withStripeMaxSize(getOrcWriterMaxStripeSize(session))
.withStripeMaxRowCount(getOrcWriterMaxStripeRows(session))
.withDictionaryMaxMemory(getOrcWriterMaxDictionaryMemory(session))
.withMaxStringStatisticsLimit(getOrcStringStatisticsLimit(session)),
.withMaxStringStatisticsLimit(stringStatisticsLimit),
IntStream.range(0, fileColumnNames.size()).toArray(),
ImmutableMap.<String, String>builder()
.put(PRESTO_VERSION_NAME, nodeVersion.toString())
Expand Down
5 changes: 4 additions & 1 deletion plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergParquetFileWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,11 +36,13 @@ public class IcebergParquetFileWriter
extends ParquetFileWriter
implements IcebergFileWriter
{
private final MetricsConfig metricsConfig;
private final Path outputPath;
private final HdfsEnvironment hdfsEnvironment;
private final HdfsContext hdfsContext;

public IcebergParquetFileWriter(
MetricsConfig metricsConfig,
OutputStream outputStream,
Callable<Void> rollbackAction,
List<Type> fileColumnTypes,
Expand All @@ -63,6 +65,7 @@ public IcebergParquetFileWriter(
fileInputColumnIndexes,
compressionCodecName,
trinoVersion);
this.metricsConfig = requireNonNull(metricsConfig, "metricsConfig is null");
this.outputPath = requireNonNull(outputPath, "outputPath is null");
this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
this.hdfsContext = requireNonNull(hdfsContext, "hdfsContext is null");
Expand All @@ -71,6 +74,6 @@ public IcebergParquetFileWriter(
@Override
public Metrics getMetrics()
{
return hdfsEnvironment.doAs(hdfsContext.getIdentity(), () -> ParquetUtil.fileMetrics(new HdfsInputFile(outputPath, hdfsEnvironment, hdfsContext), MetricsConfig.getDefault()));
return hdfsEnvironment.doAs(hdfsContext.getIdentity(), () -> ParquetUtil.fileMetrics(new HdfsInputFile(outputPath, hdfsEnvironment, hdfsContext), metricsConfig));
}
}