-
Notifications
You must be signed in to change notification settings - Fork 3k
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.
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
Implement predicate push down for parquet dereference column in Iceberg #17133
Merged
raunaqmorarka
merged 2 commits into
trinodb:master
from
leetcode-1533:iceberg_dereferenceparquet_2
Jan 18, 2024
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
...rc/test/java/io/trino/plugin/iceberg/TestIcebergParquetComplexTypesPredicatePushDown.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.trino.plugin.iceberg; | ||
|
|
||
| import io.trino.Session; | ||
| import io.trino.testing.BaseComplexTypesPredicatePushDownTest; | ||
| import io.trino.testing.QueryRunner; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import static io.trino.plugin.iceberg.IcebergTestUtils.withSmallRowGroups; | ||
| import static io.trino.testing.TestingNames.randomNameSuffix; | ||
|
|
||
| public class TestIcebergParquetComplexTypesPredicatePushDown | ||
| extends BaseComplexTypesPredicatePushDownTest | ||
| { | ||
| @Override | ||
| protected QueryRunner createQueryRunner() | ||
| throws Exception | ||
| { | ||
| return IcebergQueryRunner.builder() | ||
| .addIcebergProperty("iceberg.file-format", "PARQUET") | ||
| .build(); | ||
| } | ||
|
|
||
| @Override | ||
| protected final Session getSession() | ||
| { | ||
| return withSmallRowGroups(super.getSession()); | ||
| } | ||
|
|
||
| // The Iceberg table scan differs from Hive in that the coordinator also uses file statistics when generating the splits . | ||
| // As a result, if the predicates fall outside the bounds of the file statistics, | ||
| // the split is not created for the worker and worker won't call getParquetTupleDomain(). | ||
| // The test increased the number of row groups and introduced predicates that are within the file statistics but outside the statistics of the row groups. | ||
| @Test | ||
| public void testIcebergParquetRowTypeRowGroupPruning() | ||
| { | ||
| String tableName = "test_nested_column_pruning_" + randomNameSuffix(); | ||
| assertUpdate("CREATE TABLE " + tableName + " (col1Row ROW(a BIGINT, b BIGINT), col2 BIGINT) WITH (sorted_by=ARRAY['col2'])"); | ||
| assertUpdate("INSERT INTO " + tableName + " SELECT * FROM unnest(transform(SEQUENCE(1, 10000), x -> ROW(ROW(x*2, 100), x)))", 10000); | ||
|
|
||
| // col1Row.a only contains even numbers, in the range of [2, 20000]. | ||
| // The test has roughly 50 rows per row group due to withSmallRowGroups, [2, 100], [102, 200], ... [19902, 20000] | ||
| // 101 is a value between [2, 20000] but is an odd number, so won't be discarded by Iceberg table's statistics. | ||
| // At the same time, 101 is not within the bound of any row group. So can be discarded by Parquet's row group statistics. | ||
| assertNoDataRead("SELECT * FROM " + tableName + " WHERE col1Row.a = 101"); | ||
| assertNoDataRead("SELECT * FROM " + tableName + " WHERE col1Row.a IS NULL"); | ||
|
|
||
| assertUpdate("DROP TABLE " + tableName); | ||
| } | ||
| } |
216 changes: 216 additions & 0 deletions
216
plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestParquetPredicates.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,216 @@ | ||
| /* | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.trino.plugin.iceberg; | ||
|
|
||
| import com.google.common.collect.ImmutableList; | ||
| import com.google.common.collect.ImmutableMap; | ||
| import io.trino.spi.predicate.Domain; | ||
| import io.trino.spi.predicate.TupleDomain; | ||
| import io.trino.spi.type.RowType; | ||
| import org.apache.parquet.column.ColumnDescriptor; | ||
| import org.apache.parquet.schema.GroupType; | ||
| import org.apache.parquet.schema.MessageType; | ||
| import org.apache.parquet.schema.PrimitiveType; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
|
|
||
| import static io.trino.parquet.ParquetTypeUtils.getDescriptors; | ||
| import static io.trino.plugin.iceberg.ColumnIdentity.TypeCategory.PRIMITIVE; | ||
| import static io.trino.plugin.iceberg.ColumnIdentity.TypeCategory.STRUCT; | ||
| import static io.trino.plugin.iceberg.IcebergPageSourceProvider.getParquetTupleDomain; | ||
| import static io.trino.spi.predicate.TupleDomain.withColumnDomains; | ||
| import static io.trino.spi.type.IntegerType.INTEGER; | ||
| import static io.trino.spi.type.RowType.rowType; | ||
| import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; | ||
| import static org.apache.parquet.schema.Type.Repetition.OPTIONAL; | ||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| public class TestParquetPredicates | ||
| { | ||
| @Test | ||
| public void testParquetTupleDomainStructWithPrimitiveColumnPredicate() | ||
| { | ||
| // trino type | ||
| RowType baseType = rowType( | ||
| RowType.field("a", INTEGER), | ||
| RowType.field("b", INTEGER), | ||
| RowType.field("c", INTEGER)); | ||
|
|
||
| // iceberg type | ||
| ColumnIdentity fieldA = new ColumnIdentity(1, "a", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldB = new ColumnIdentity(2, "b", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldC = new ColumnIdentity(3, "c", PRIMITIVE, ImmutableList.of()); | ||
|
|
||
| // parquet type | ||
| MessageType fileSchema = new MessageType("iceberg_schema", | ||
| new GroupType(OPTIONAL, "row_field", | ||
| new PrimitiveType(OPTIONAL, INT32, "a").withId(1), | ||
| new PrimitiveType(OPTIONAL, INT32, "b").withId(2), | ||
| new PrimitiveType(OPTIONAL, INT32, "c").withId(3))); | ||
|
|
||
| // predicate domain | ||
| IcebergColumnHandle projectedColumn = new IcebergColumnHandle( | ||
| new ColumnIdentity( | ||
| 5, | ||
| "row_field", | ||
| STRUCT, | ||
| ImmutableList.of(fieldA, fieldB, fieldC)), | ||
| baseType, | ||
| ImmutableList.of(2), | ||
| INTEGER, | ||
| Optional.empty()); | ||
| Domain predicateDomain = Domain.singleValue(INTEGER, 123L); | ||
| TupleDomain<IcebergColumnHandle> tupleDomain = withColumnDomains(ImmutableMap.of(projectedColumn, predicateDomain)); | ||
|
|
||
| Map<List<String>, ColumnDescriptor> descriptorsByPath = getDescriptors(fileSchema, fileSchema); | ||
| TupleDomain<ColumnDescriptor> calculatedTupleDomain = getParquetTupleDomain(descriptorsByPath, tupleDomain); | ||
|
|
||
| assertThat(calculatedTupleDomain.getDomains().orElseThrow().size()).isEqualTo(1); | ||
| ColumnDescriptor selectedColumnDescriptor = descriptorsByPath.get(ImmutableList.of("row_field", "b")); | ||
| assertThat(calculatedTupleDomain.getDomains().orElseThrow().get(selectedColumnDescriptor)).isEqualTo(predicateDomain); | ||
| } | ||
|
|
||
| @Test | ||
| public void testParquetTupleDomainStructWithPrimitiveColumnDifferentIdPredicate() | ||
| { | ||
| // trino type | ||
| RowType baseType = rowType( | ||
| RowType.field("a", INTEGER), | ||
| RowType.field("b", INTEGER), | ||
| RowType.field("c", INTEGER)); | ||
|
|
||
| // iceberg type | ||
| ColumnIdentity fieldA = new ColumnIdentity(1, "a", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldB = new ColumnIdentity(2, "b", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldC = new ColumnIdentity(4, "c", PRIMITIVE, ImmutableList.of()); | ||
|
|
||
| // parquet type | ||
| MessageType fileSchema = new MessageType("iceberg_schema", | ||
| new GroupType(OPTIONAL, "row_field", | ||
| new PrimitiveType(OPTIONAL, INT32, "a").withId(1), | ||
| new PrimitiveType(OPTIONAL, INT32, "b").withId(2), | ||
| new PrimitiveType(OPTIONAL, INT32, "c").withId(3)).withId(5)); | ||
|
|
||
| // predicate domain | ||
| IcebergColumnHandle projectedColumn = new IcebergColumnHandle( | ||
| new ColumnIdentity( | ||
| 5, | ||
| "row_field", | ||
| STRUCT, | ||
| ImmutableList.of(fieldA, fieldB, fieldC)), | ||
| baseType, | ||
| ImmutableList.of(4), | ||
| INTEGER, | ||
| Optional.empty()); | ||
| Domain predicateDomain = Domain.singleValue(INTEGER, 123L); | ||
| TupleDomain<IcebergColumnHandle> tupleDomain = withColumnDomains(ImmutableMap.of(projectedColumn, predicateDomain)); | ||
|
|
||
| Map<List<String>, ColumnDescriptor> descriptorsByPath = getDescriptors(fileSchema, fileSchema); | ||
| TupleDomain<ColumnDescriptor> calculatedTupleDomain = getParquetTupleDomain(descriptorsByPath, tupleDomain); | ||
| // same name but different Id between iceberg and parquet for field c | ||
| assertThat(calculatedTupleDomain.isAll()).isTrue(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testParquetTupleDomainStructWithComplexColumnPredicate() | ||
| { | ||
| // trino type | ||
| RowType nestedType = rowType( | ||
| RowType.field("c1", INTEGER), | ||
| RowType.field("c2", INTEGER)); | ||
| RowType baseType = rowType( | ||
| RowType.field("a", INTEGER), | ||
| RowType.field("b", INTEGER), | ||
| RowType.field("c", nestedType)); | ||
|
|
||
| // iceberg type | ||
| ColumnIdentity fieldC11 = new ColumnIdentity(1, "c1", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldC12 = new ColumnIdentity(2, "c2", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldA = new ColumnIdentity(3, "a", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldB = new ColumnIdentity(4, "b", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldC = new ColumnIdentity(5, "c", STRUCT, ImmutableList.of(fieldC11, fieldC12)); | ||
|
|
||
| // parquet type | ||
| MessageType fileSchema = new MessageType("iceberg_schema", | ||
| new GroupType(OPTIONAL, "row_field", | ||
| new PrimitiveType(OPTIONAL, INT32, "a").withId(3), | ||
| new PrimitiveType(OPTIONAL, INT32, "b").withId(4), | ||
| new GroupType(OPTIONAL, | ||
| "c", | ||
| new PrimitiveType(OPTIONAL, INT32, "c1").withId(1), | ||
| new PrimitiveType(OPTIONAL, INT32, "c2").withId(2)).withId(5))); | ||
| // predicate domain | ||
| IcebergColumnHandle projectedColumn = new IcebergColumnHandle( | ||
| new ColumnIdentity( | ||
| 6, | ||
| "row_field", | ||
| STRUCT, | ||
| ImmutableList.of(fieldA, fieldB, fieldC)), | ||
| baseType, | ||
| ImmutableList.of(5), | ||
| nestedType, | ||
| Optional.empty()); | ||
|
|
||
| Domain predicateDomain = Domain.onlyNull(nestedType); | ||
| TupleDomain<IcebergColumnHandle> tupleDomain = withColumnDomains(ImmutableMap.of(projectedColumn, predicateDomain)); | ||
|
|
||
| Map<List<String>, ColumnDescriptor> descriptorsByPath = getDescriptors(fileSchema, fileSchema); | ||
| TupleDomain<ColumnDescriptor> calculatedTupleDomain = getParquetTupleDomain(descriptorsByPath, tupleDomain); | ||
|
|
||
| assertThat(calculatedTupleDomain.isAll()).isTrue(); | ||
| } | ||
|
|
||
| @Test | ||
| public void testParquetTupleDomainStructWithMissingPrimitiveColumn() | ||
| { | ||
| // trino type | ||
| RowType baseType = rowType( | ||
| RowType.field("a", INTEGER), | ||
| RowType.field("b", INTEGER), | ||
| RowType.field("missing", INTEGER)); | ||
|
|
||
| // iceberg type | ||
| ColumnIdentity fieldA = new ColumnIdentity(1, "a", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldB = new ColumnIdentity(2, "b", PRIMITIVE, ImmutableList.of()); | ||
| ColumnIdentity fieldC = new ColumnIdentity(3, "missing", PRIMITIVE, ImmutableList.of()); | ||
|
|
||
| // parquet type | ||
| MessageType fileSchema = new MessageType("iceberg_schema", | ||
| new GroupType(OPTIONAL, "row_field", | ||
| new PrimitiveType(OPTIONAL, INT32, "a").withId(1), | ||
| new PrimitiveType(OPTIONAL, INT32, "b").withId(2))); | ||
|
|
||
| // predicate domain | ||
| IcebergColumnHandle projectedColumn = new IcebergColumnHandle( | ||
| new ColumnIdentity( | ||
| 5, | ||
| "row_field", | ||
| STRUCT, | ||
| ImmutableList.of(fieldA, fieldB, fieldC)), | ||
| baseType, | ||
| ImmutableList.of(3), | ||
| INTEGER, | ||
| Optional.empty()); | ||
| Domain predicateDomain = Domain.singleValue(INTEGER, 123L); | ||
| TupleDomain<IcebergColumnHandle> tupleDomain = withColumnDomains(ImmutableMap.of(projectedColumn, predicateDomain)); | ||
|
|
||
| Map<List<String>, ColumnDescriptor> descriptorsByPath = getDescriptors(fileSchema, fileSchema); | ||
| TupleDomain<ColumnDescriptor> calculatedTupleDomain = getParquetTupleDomain(descriptorsByPath, tupleDomain); | ||
|
|
||
| assertThat(calculatedTupleDomain.isAll()).isTrue(); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why 5 ?