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

Add Parquet bloom filter write support to Iceberg connector #21602

Merged
merged 2 commits into from
Jun 24, 2024
Merged

Add Parquet bloom filter write support to Iceberg connector #21602

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
cd69afd
Add Parquet Bloom filter write support to Iceberg connector
jkylling Apr 18, 2024
ac3b281
List unhealthy containers when product tests health check fails
jkylling May 3, 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
4 changes: 4 additions & 0 deletions docs/src/main/sphinx/connector/iceberg.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -716,6 +716,10 @@ connector using a {doc}`WITH </sql/create-table-as>` clause.
* - `orc_bloom_filter_fpp`
- The ORC bloom filters false positive probability. Requires ORC format.
Defaults to `0.05`.
* - `parquet_bloom_filter_columns`
- Comma-separated list of columns to use for Parquet bloom filter. It improves
the performance of queries using Equality and IN predicates when reading
Parquet files. Requires Parquet format. Defaults to `[]`.
:::

The table definition below specifies to use Parquet files, partitioning by columns
Expand Down
9 changes: 6 additions & 3 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 @@ -73,6 +73,7 @@
import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_FPP_PROPERTY;
import static io.trino.plugin.iceberg.IcebergUtil.getOrcBloomFilterColumns;
import static io.trino.plugin.iceberg.IcebergUtil.getOrcBloomFilterFpp;
import static io.trino.plugin.iceberg.IcebergUtil.getParquetBloomFilterColumns;
import static io.trino.plugin.iceberg.TypeConverter.toTrinoType;
import static io.trino.plugin.iceberg.util.OrcTypeConverter.toOrcType;
import static io.trino.plugin.iceberg.util.PrimitiveTypeMapBuilder.makeTypeMap;
Expand Down Expand Up @@ -126,7 +127,7 @@ public IcebergFileWriter createDataFileWriter(
{
return switch (fileFormat) {
// TODO use metricsConfig https://github.com/trinodb/trino/issues/9791
case PARQUET -> createParquetWriter(MetricsConfig.getDefault(), fileSystem, outputPath, icebergSchema, session);
case PARQUET -> createParquetWriter(MetricsConfig.getDefault(), fileSystem, outputPath, icebergSchema, session, storageProperties);
case ORC -> createOrcWriter(metricsConfig, fileSystem, outputPath, icebergSchema, session, storageProperties, getOrcStringStatisticsLimit(session));
case AVRO -> createAvroWriter(fileSystem, outputPath, icebergSchema, session);
};
Expand All @@ -140,7 +141,7 @@ public IcebergFileWriter createPositionDeleteWriter(
Map<String, String> storageProperties)
{
return switch (fileFormat) {
case PARQUET -> createParquetWriter(FULL_METRICS_CONFIG, fileSystem, outputPath, POSITION_DELETE_SCHEMA, session);
case PARQUET -> createParquetWriter(FULL_METRICS_CONFIG, fileSystem, outputPath, POSITION_DELETE_SCHEMA, session, storageProperties);
case ORC -> createOrcWriter(FULL_METRICS_CONFIG, fileSystem, outputPath, POSITION_DELETE_SCHEMA, session, storageProperties, DataSize.ofBytes(Integer.MAX_VALUE));
case AVRO -> createAvroWriter(fileSystem, outputPath, POSITION_DELETE_SCHEMA, session);
};
Expand All @@ -151,7 +152,8 @@ private IcebergFileWriter createParquetWriter(
TrinoFileSystem fileSystem,
Location outputPath,
Schema icebergSchema,
ConnectorSession session)
ConnectorSession session,
Map<String, String> storageProperties)
{
List<String> fileColumnNames = icebergSchema.columns().stream()
.map(Types.NestedField::name)
Expand All @@ -170,6 +172,7 @@ private IcebergFileWriter createParquetWriter(
.setMaxPageValueCount(getParquetWriterPageValueCount(session))
.setMaxBlockSize(getParquetWriterBlockSize(session))
.setBatchSize(getParquetWriterBatchSize(session))
.setBloomFilterColumns(getParquetBloomFilterColumns(storageProperties))
Copy link
Contributor

Choose a reason for hiding this comment

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

a few lines above you have the original fileColumnNames - please use those in correlation with what is specified in the table properties (case insensitive name matching) in getParquetBloomFilterColumns.

Also a new test to add: schema evolution - create a table with a bunch of bloom filter columns, drop one of the columns which was specified as bloom filter column and make sure that you don't get any errors . I'm guessing we'd have to filter out in getParquetBloomFilterColumns the column names which don't exist anymore.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The writing logic ignores non-existent columns for which the Bloom filter property is set.

.build();

HiveCompressionCodec hiveCompressionCodec = getCompressionCodec(session);
Expand Down
19 changes: 19 additions & 0 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ public class IcebergTableProperties
public static final String FORMAT_VERSION_PROPERTY = "format_version";
public static final String ORC_BLOOM_FILTER_COLUMNS_PROPERTY = "orc_bloom_filter_columns";
public static final String ORC_BLOOM_FILTER_FPP_PROPERTY = "orc_bloom_filter_fpp";
public static final String PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY = "parquet_bloom_filter_columns";
Copy link
Contributor

Choose a reason for hiding this comment

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

spark-sql (default)> CREATE TABLE t1 (testInteger INTEGER, testLong BIGINT, testString STRING, testDouble DOUBLE, testFloat REAL)
                   > USING iceberg
                   > TBLPROPERTIES (
                   > 'write.parquet.bloom-filter-enabled.column.testInteger' = true, 
                   > 'write.parquet.bloom-filter-enabled.column.testLong' = true,
                   > 'write.parquet.bloom-filter-enabled.column.testString' = true,
                   > 'write.parquet.bloom-filter-enabled.column.testDouble' = true,
                   > 'write.parquet.bloom-filter-enabled.column.testFloat' = true
                   > );

trino> show create table iceberg.default.t1;
                           Create Table                           
------------------------------------------------------------------
 CREATE TABLE iceberg.default.t1 (                                
    testinteger integer,                                          
    testlong bigint,                                              
    teststring varchar,                                           
    testdouble double,                                            
    testfloat real                                                
 )                                                                
 WITH (                                                           
    format = 'PARQUET',                                           
    format_version = 2,                                           
    location = 'hdfs://hadoop-master:9000/user/hive/warehouse/t1' 
 )                                                                
(1 row)

Shouldn't we see in SHOW CREATE TABLE the bloom filter columns now that we're dealing with a supported table property?

Copy link
Contributor

Choose a reason for hiding this comment

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

Modify io.trino.plugin.iceberg.IcebergUtil#getIcebergTableProperties

Copy link
Contributor

Choose a reason for hiding this comment

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

i tried on the above scaffolding

SELECT COUNT(*) FROM iceberg.default.t1 where testInteger in (9444, -88777, 6711111);

and see the following

"queryStats" : {
  ....
    "physicalInputDataSize" : "656400B",
    "failedPhysicalInputDataSize" : "0B",
    "physicalInputPositions" : 5,

This seems not to overlap with the expectations from io.trino.testing.BaseTestParquetWithBloomFilters#testBloomFilterRowGroupPruning(io.trino.spi.connector.CatalogSchemaTableName, java.lang.String)

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We could add a toLowerCase to getParquetBloomFilterColumns to handle this? It looks like we have the same issues for the Iceberg ORC Bloom filters. Should we handle case sensitivity in this PR, or handle it in a follow up?

Copy link
Contributor

Choose a reason for hiding this comment

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

let's rather fix the functionality in the existing PR instead of delivering a half-baked functionality which may potentially back-fire with bugs.

Copy link
Contributor

Choose a reason for hiding this comment

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

An alternative with less headaches would be to register a pre-created resource table and check the query stats on it similar to what has been done on https://github.com/trinodb/trino/blob/ca209630136eabda2449594ef2b6a4d82fb9c2e5/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergReadVersionedTableByTemporal.java

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Easy access to this would be useful to have in the product tests. It would allow the product tests in this PR to give more coverage. Unfortunately, product tests are not my cup of tea for Friday hacking 😅

Copy link
Contributor

Choose a reason for hiding this comment

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

We need a mechanism to get the query stats in the product tests to ensure that the bloom filter is actually effective and we don't introduce while refactoring regressions.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Would someone be able to help add this logic? I don't have much experience with the product tests and unfortunately don't have much capacity to follow up on this at the moment. It would be much appreciated!

Copy link
Member

Choose a reason for hiding this comment

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

@findinpath aren't we already testing effectiveness of bloom filter in query runner tests ? I'm not sure that we should block this PR over checking this in product tests as well, we don't do that with Apache Hive for bloom filters in hive connector as well.


private final List<PropertyMetadata<?>> tableProperties;

Expand Down Expand Up @@ -107,6 +108,18 @@ public IcebergTableProperties(
orcWriterConfig.getDefaultBloomFilterFpp(),
IcebergTableProperties::validateOrcBloomFilterFpp,
false))
.add(new PropertyMetadata<>(
PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY,
"Parquet Bloom filter index columns",
new ArrayType(VARCHAR),
List.class,
ImmutableList.of(),
false,
value -> ((List<?>) value).stream()
.map(String.class::cast)
.map(name -> name.toLowerCase(ENGLISH))
.collect(toImmutableList()),
value -> value))
.build();
}

Expand Down Expand Up @@ -169,4 +182,10 @@ private static void validateOrcBloomFilterFpp(double fpp)
throw new TrinoException(INVALID_TABLE_PROPERTY, "Bloom filter fpp value must be between 0.0 and 1.0");
}
}

public static List<String> getParquetBloomFilterColumns(Map<String, Object> tableProperties)
{
List<String> parquetBloomFilterColumns = (List<String>) tableProperties.get(PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY);
return parquetBloomFilterColumns == null ? ImmutableList.of() : ImmutableList.copyOf(parquetBloomFilterColumns);
}
}
50 changes: 46 additions & 4 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static com.google.common.collect.Iterables.getOnlyElement;
import static io.airlift.slice.Slices.utf8Slice;
import static io.trino.parquet.writer.ParquetWriter.SUPPORTED_BLOOM_FILTER_TYPES;
import static io.trino.plugin.base.io.ByteBuffers.getWrappedBytes;
import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
import static io.trino.plugin.iceberg.ColumnIdentity.createColumnIdentity;
Expand All @@ -117,6 +118,7 @@
import static io.trino.plugin.iceberg.IcebergTableProperties.LOCATION_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_COLUMNS_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_FPP_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.PARTITIONING_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.SORTED_BY_PROPERTY;
import static io.trino.plugin.iceberg.IcebergTableProperties.getPartitioning;
Expand Down Expand Up @@ -166,6 +168,7 @@
import static org.apache.iceberg.TableProperties.OBJECT_STORE_PATH;
import static org.apache.iceberg.TableProperties.ORC_BLOOM_FILTER_COLUMNS;
import static org.apache.iceberg.TableProperties.ORC_BLOOM_FILTER_FPP;
import static org.apache.iceberg.TableProperties.PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX;
import static org.apache.iceberg.TableProperties.WRITE_DATA_LOCATION;
import static org.apache.iceberg.TableProperties.WRITE_LOCATION_PROVIDER_IMPL;
import static org.apache.iceberg.TableProperties.WRITE_METADATA_LOCATION;
Expand Down Expand Up @@ -264,6 +267,12 @@ public static Map<String, Object> getIcebergTableProperties(Table icebergTable)
properties.put(ORC_BLOOM_FILTER_FPP_PROPERTY, Double.parseDouble(orcBloomFilterFpp.get()));
}

// iceberg Parquet format bloom filter properties
Set<String> parquetBloomFilterColumns = getParquetBloomFilterColumns(icebergTable.properties());
if (!parquetBloomFilterColumns.isEmpty()) {
properties.put(PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY, ImmutableList.copyOf(parquetBloomFilterColumns));
}

return properties.buildOrThrow();
}

Expand All @@ -280,6 +289,14 @@ public static Optional<String> getOrcBloomFilterColumns(Map<String, String> prop
return orcBloomFilterColumns;
}

public static Set<String> getParquetBloomFilterColumns(Map<String, String> properties)
{
return properties.entrySet().stream()
.filter(entry -> entry.getKey().startsWith(PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX) && "true".equals(entry.getValue()))
.map(entry -> entry.getKey().substring(PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX.length()))
Copy link
Contributor

@findinpath findinpath May 3, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Do we need to lowercase the column names?
We'd probably need a spark compatibility test using case sensitive column names to check this
I see already testSparkReadingTrinoBloomFilters

.collect(toImmutableSet());
}

public static Optional<String> getOrcBloomFilterFpp(Map<String, String> properties)
{
return Stream.of(
Expand Down Expand Up @@ -733,14 +750,24 @@ public static Map<String, String> createTableProperties(ConnectorTableMetadata t
propertiesBuilder.put(FORMAT_VERSION, Integer.toString(IcebergTableProperties.getFormatVersion(tableMetadata.getProperties())));

// iceberg ORC format bloom filter properties used by create table
List<String> columns = IcebergTableProperties.getOrcBloomFilterColumns(tableMetadata.getProperties());
if (!columns.isEmpty()) {
List<String> orcBloomFilterColumns = IcebergTableProperties.getOrcBloomFilterColumns(tableMetadata.getProperties());
if (!orcBloomFilterColumns.isEmpty()) {
checkFormatForProperty(fileFormat.toIceberg(), FileFormat.ORC, ORC_BLOOM_FILTER_COLUMNS_PROPERTY);
validateOrcBloomFilterColumns(tableMetadata, columns);
propertiesBuilder.put(ORC_BLOOM_FILTER_COLUMNS, Joiner.on(",").join(columns));
validateOrcBloomFilterColumns(tableMetadata, orcBloomFilterColumns);
propertiesBuilder.put(ORC_BLOOM_FILTER_COLUMNS, Joiner.on(",").join(orcBloomFilterColumns));
propertiesBuilder.put(ORC_BLOOM_FILTER_FPP, String.valueOf(IcebergTableProperties.getOrcBloomFilterFpp(tableMetadata.getProperties())));
}

// iceberg Parquet format bloom filter properties used by create table
List<String> parquetBloomFilterColumns = IcebergTableProperties.getParquetBloomFilterColumns(tableMetadata.getProperties());
if (!parquetBloomFilterColumns.isEmpty()) {
checkFormatForProperty(fileFormat.toIceberg(), FileFormat.PARQUET, PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY);
validateParquetBloomFilterColumns(tableMetadata, parquetBloomFilterColumns);
for (String column : parquetBloomFilterColumns) {
propertiesBuilder.put(PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIX + column, "true");
}
}

if (tableMetadata.getComment().isPresent()) {
propertiesBuilder.put(TABLE_COMMENT, tableMetadata.getComment().get());
}
Expand Down Expand Up @@ -841,6 +868,21 @@ private static void validateOrcBloomFilterColumns(ConnectorTableMetadata tableMe
}
}

private static void validateParquetBloomFilterColumns(ConnectorTableMetadata tableMetadata, List<String> parquetBloomFilterColumns)
{
Map<String, Type> columnTypes = tableMetadata.getColumns().stream()
.collect(toImmutableMap(ColumnMetadata::getName, ColumnMetadata::getType));
for (String column : parquetBloomFilterColumns) {
Type type = columnTypes.get(column);
if (type == null) {
throw new TrinoException(INVALID_TABLE_PROPERTY, format("Parquet Bloom filter column %s not present in schema", column));
}
if (!SUPPORTED_BLOOM_FILTER_TYPES.contains(type)) {
throw new TrinoException(INVALID_TABLE_PROPERTY, format("Parquet Bloom filter column %s has unsupported type %s", column, type.getDisplayName()));
}
}
}

public static int parseVersion(String metadataFileName)
throws TrinoException
{
Expand Down
47 changes: 28 additions & 19 deletions ...ino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergParquetWithBloomFilters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,18 +14,20 @@
package io.trino.plugin.iceberg;

import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import io.trino.plugin.hive.TestingHivePlugin;
import io.trino.spi.connector.CatalogSchemaTableName;
import io.trino.spi.connector.SchemaTableName;
import io.trino.testing.BaseTestParquetWithBloomFilters;
import io.trino.testing.MaterializedResult;
import io.trino.testing.QueryRunner;
import org.junit.jupiter.api.Test;

import java.nio.file.Path;
import java.util.List;

import static io.trino.testing.MaterializedResult.resultBuilder;
import static io.trino.testing.QueryAssertions.assertContains;
import static io.trino.testing.TestingNames.randomNameSuffix;
import static java.lang.String.format;
import static org.assertj.core.api.Assertions.assertThat;

public class TestIcebergParquetWithBloomFilters
extends BaseTestParquetWithBloomFilters
Expand All @@ -34,30 +36,37 @@ public class TestIcebergParquetWithBloomFilters
protected QueryRunner createQueryRunner()
throws Exception
{
QueryRunner queryRunner = IcebergQueryRunner.builder().build();
Path dataDirectory = queryRunner.getCoordinator().getBaseDataDir().resolve("iceberg_data");

// create hive catalog
queryRunner.installPlugin(new TestingHivePlugin(dataDirectory));
queryRunner.createCatalog("hive", "hive", ImmutableMap.<String, String>builder()
.put("hive.security", "allow-all")
.buildOrThrow());

return queryRunner;
return IcebergQueryRunner.builder().build();
}

@Override
protected CatalogSchemaTableName createParquetTableWithBloomFilter(String columnName, List<Integer> testValues)
{
// create the managed table
String tableName = "parquet_with_bloom_filters_" + randomNameSuffix();
CatalogSchemaTableName hiveCatalogSchemaTableName = new CatalogSchemaTableName("hive", new SchemaTableName("tpch", tableName));
CatalogSchemaTableName icebergCatalogSchemaTableName = new CatalogSchemaTableName("iceberg", new SchemaTableName("tpch", tableName));
assertUpdate(format("CREATE TABLE %s WITH (format = 'PARQUET', parquet_bloom_filter_columns = ARRAY['%s']) AS SELECT * FROM (VALUES %s) t(%s)", hiveCatalogSchemaTableName, columnName, Joiner.on(", ").join(testValues), columnName), testValues.size());
CatalogSchemaTableName catalogSchemaTableName = new CatalogSchemaTableName("iceberg", new SchemaTableName("tpch", tableName));
assertUpdate(format("CREATE TABLE %s WITH (format = 'PARQUET', parquet_bloom_filter_columns = ARRAY['%s']) AS SELECT * FROM (VALUES %s) t(%s)", catalogSchemaTableName, columnName, Joiner.on(", ").join(testValues), columnName), testValues.size());

return catalogSchemaTableName;
}

@Test
public void testBloomFilterPropertiesArePersistedDuringCreate()
{
String tableName = "test_metadata_write_properties_" + randomNameSuffix();
assertQuerySucceeds("CREATE TABLE " + tableName + " (A bigint, b bigint, c bigint) WITH (" +
"format = 'parquet'," +
"parquet_bloom_filter_columns = array['a','B'])");

// migrate the hive table to the iceberg table
assertUpdate("CALL iceberg.system.migrate('tpch', '" + tableName + "', 'false')");
MaterializedResult actualProperties = computeActual("SELECT * FROM \"" + tableName + "$properties\"");
assertThat(actualProperties).isNotNull();
MaterializedResult expectedProperties = resultBuilder(getSession())
.row("write.parquet.bloom-filter-enabled.column.a", "true")
.row("write.parquet.bloom-filter-enabled.column.b", "true")
.build();
assertContains(actualProperties, expectedProperties);

return icebergCatalogSchemaTableName;
assertThat((String) computeScalar("SHOW CREATE TABLE " + tableName))
.contains("parquet_bloom_filter_columns");
}
}
Loading