Skip to content

Add support for struct field based filtering #123

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
rdblue merged 22 commits into from
May 28, 2019
Merged

Add support for struct field based filtering #123

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
a48e9d9
Add support for struct field based filtering
prodeezy Mar 7, 2019
0c9c631
throw error when struct field not found and consolidate logic for fin…
prodeezy Mar 11, 2019
a5e4d16
Fixed exception error, reused variable in bind
prodeezy Mar 13, 2019
dae1922
Adding unit tests to skip based on struct fields on dictionary and me…
prodeezy Mar 14, 2019
cdda798
Merge branch 'master' into issue-122-support-struct-filter-expressions
prodeezy Mar 22, 2019
f35e3d6
Added accessor based fetching of fields in BoundReference, build inde…
prodeezy Mar 25, 2019
71aaba6
Added caseInsensitiveFindField and struct based unit tests for Evaluator
prodeezy Mar 26, 2019
9f72f36
Fixed tests to create struct schema properly
prodeezy Mar 26, 2019
1097b11
made id to accessor map load lazily and cached reference
prodeezy Mar 26, 2019
8fa0a11
Merge branch 'master' into issue-122-support-struct-filter-expressions
prodeezy Apr 10, 2019
6438da5
addressed style comments
prodeezy Apr 10, 2019
deb5c59
Merge branch 'master' into issue-122-support-struct-filter-expressions
prodeezy May 9, 2019
f7de4e2
Moved accessor code out of BoundReference into package private classes
prodeezy May 10, 2019
1deff5b
Moved Field to Accessor mapping to Schema
prodeezy May 11, 2019
b1f59e5
Remove extra lines
prodeezy May 14, 2019
cbb3a9f
Using accessor to fetch field position
prodeezy May 14, 2019
25f6f91
Cleared out code from BoundReference
prodeezy May 14, 2019
b699257
toString prings id and accessor-type
prodeezy May 24, 2019
f3f1783
Merge branch 'master' into issue-122-support-struct-filter-expressions
prodeezy May 24, 2019
a88603e
Minor updates for style and to minimize additions to the public API.
rdblue May 25, 2019
83227b0
Merge pull request #1 from rdblue/pr-123-support-filter-on-nested-fields
prodeezy May 25, 2019
2031eb5
Fix broken test
prodeezy May 28, 2019
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
33 changes: 27 additions & 6 deletions api/src/main/java/com/netflix/iceberg/expressions/BoundReference.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
package com.netflix.iceberg.expressions;

import com.netflix.iceberg.StructLike;
import com.netflix.iceberg.exceptions.ValidationException;
import com.netflix.iceberg.types.Type;
import com.netflix.iceberg.types.Types;
import java.util.List;
Expand All @@ -32,19 +31,41 @@ public class BoundReference<T> implements Reference {

BoundReference(Types.StructType struct, int fieldId) {
this.fieldId = fieldId;
this.pos = find(fieldId, struct);
this.type = struct.fields().get(pos).type();
this.pos = findTopFieldPos(fieldId, struct);

@xabriel xabriel Mar 11, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

What does it mean to have a pos = -1?

I ask because although this field is private, we do expose it via pos() and toString() methods, so we may need to populate this field with the actual position of the matched inner struct.

@prodeezy prodeezy Mar 14, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yeah, good point. going to write some unit tests to evaluate impact. will handle accordingly.

@prodeezy prodeezy Mar 26, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

handled this using accessors

this.type = (pos > -1) ? struct.fields().get(pos).type() : findStructFieldType(fieldId, struct);
Comment thread
prodeezy marked this conversation as resolved.
Outdated
}

private int find(int fieldId, Types.StructType struct) {
private int findTopFieldPos(int fieldId, Types.StructType struct) {
List<Types.NestedField> fields = struct.fields();
for (int i = 0; i < fields.size(); i += 1) {
if (fields.get(i).fieldId() == fieldId) {
return i;
}
}
throw new ValidationException(
"Cannot find top-level field id %d in struct: %s", fieldId, struct);

return -1;
}

private Type findStructFieldType(int fieldId, Types.StructType struct) {
List<Types.NestedField> fields = struct.fields();
for (int i = 0; i < fields.size(); i += 1) {

if (fields.get(i).fieldId() == fieldId) {
return fields.get(i).type();
} else {

// look under struct by calling this method recursively
if (fields.get(i).type() instanceof Types.StructType) {
Types.StructType subStruct = fields.get(i).type().asStructType();

Type ret = findStructFieldType(fieldId, subStruct);
if (ret != null) {
Comment thread
prodeezy marked this conversation as resolved.
Outdated
return ret;
}
}
}
}
return null;

@fbocse fbocse Mar 8, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Should this method be returning a null value in case none of the nested fields of this struct match by field id?
Instead of an instance of BoundReference with a null private attribute type would it be acceptable that we instead throw new ValidationException("Cannot find nested field id %d in struct: %s", fieldId, struct); ?

@prodeezy prodeezy Mar 11, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

done.

}

public Type type() {
Expand Down
26 changes: 6 additions & 20 deletions api/src/main/java/com/netflix/iceberg/expressions/InclusiveMetricsEvaluator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,8 +122,6 @@ public <T> Boolean isNull(BoundReference<T> ref) {
// no need to check whether the field is required because binding evaluates that case
// if the column has no null values, the expression cannot match
Integer id = ref.fieldId();
Preconditions.checkNotNull(struct.field(id),
"Cannot filter by nested column: %s", schema.findField(id));

if (nullCounts != null && nullCounts.containsKey(id) && nullCounts.get(id) == 0) {
return ROWS_CANNOT_MATCH;
Expand All @@ -137,8 +135,6 @@ public <T> Boolean notNull(BoundReference<T> ref) {
// no need to check whether the field is required because binding evaluates that case
// if the column has no non-null values, the expression cannot match
Integer id = ref.fieldId();
Preconditions.checkNotNull(struct.field(id),
"Cannot filter by nested column: %s", schema.findField(id));

if (valueCounts != null && valueCounts.containsKey(id) &&
nullCounts != null && nullCounts.containsKey(id) &&
Expand All @@ -152,11 +148,9 @@ public <T> Boolean notNull(BoundReference<T> ref) {
@Override
public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

if (lowerBounds != null && lowerBounds.containsKey(id)) {
T lower = Conversions.fromByteBuffer(field.type(), lowerBounds.get(id));
T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));

int cmp = lit.comparator().compare(lower, lit.value());
if (cmp >= 0) {
Expand All @@ -170,11 +164,9 @@ public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

if (lowerBounds != null && lowerBounds.containsKey(id)) {
T lower = Conversions.fromByteBuffer(field.type(), lowerBounds.get(id));
T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));

int cmp = lit.comparator().compare(lower, lit.value());
if (cmp > 0) {
Expand All @@ -188,11 +180,9 @@ public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

if (upperBounds != null && upperBounds.containsKey(id)) {
T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));

int cmp = lit.comparator().compare(upper, lit.value());
if (cmp <= 0) {
Expand All @@ -206,11 +196,9 @@ public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

if (upperBounds != null && upperBounds.containsKey(id)) {
T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));

int cmp = lit.comparator().compare(upper, lit.value());
if (cmp < 0) {
Expand All @@ -224,11 +212,9 @@ public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

if (lowerBounds != null && lowerBounds.containsKey(id)) {
T lower = Conversions.fromByteBuffer(struct.field(id).type(), lowerBounds.get(id));
T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));

int cmp = lit.comparator().compare(lower, lit.value());
if (cmp > 0) {
Expand All @@ -237,7 +223,7 @@ public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
}

if (upperBounds != null && upperBounds.containsKey(id)) {
T upper = Conversions.fromByteBuffer(field.type(), upperBounds.get(id));
T upper = Conversions.fromByteBuffer(ref.type(), upperBounds.get(id));

int cmp = lit.comparator().compare(upper, lit.value());
if (cmp < 0) {
Expand Down
40 changes: 35 additions & 5 deletions api/src/main/java/com/netflix/iceberg/expressions/UnboundPredicate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package com.netflix.iceberg.expressions;

import com.netflix.iceberg.exceptions.ValidationException;
import com.netflix.iceberg.types.Type;
import com.netflix.iceberg.types.Types;

import static com.netflix.iceberg.expressions.Expression.Operation.IS_NULL;
Expand Down Expand Up @@ -67,11 +68,13 @@ Expression bind(Types.StructType struct) {
*/
public Expression bind(Types.StructType struct, boolean caseSensitive) {
Types.NestedField field;
if (caseSensitive) {
field = struct.field(ref().name());
} else {
field = struct.caseInsensitiveField(ref().name());
}
String expressionFieldPath = ref().name();

boolean isStructFieldExp = expressionFieldPath.indexOf('.') > -1 &&
struct.field(expressionFieldPath.split("\\.")[0]).type() instanceof Types.StructType;
Comment thread
prodeezy marked this conversation as resolved.
Outdated

field = isStructFieldExp ? findStructField(struct, expressionFieldPath, caseSensitive) :
caseSensitive ? struct.field(ref().name()) : struct.caseInsensitiveField(ref().name());

ValidationException.check(field != null,
"Cannot find field '%s' in struct: %s", ref().name(), struct);
Expand Down Expand Up @@ -132,4 +135,31 @@ public Expression bind(Types.StructType struct, boolean caseSensitive) {
}
return new BoundPredicate<>(op(), new BoundReference<>(struct, field.fieldId()), lit);
}

private Types.NestedField findStructField(Types.StructType struct, String expressionFieldPath,
boolean caseSensitive) {

String[] structFields = expressionFieldPath.split("\\.");
String lastFieldInPath = structFields[structFields.length - 1];

Types.StructType subField = struct;
int i=0;

Type nextFieldType = caseSensitive ? subField.field(structFields[i]).type() :
subField.caseInsensitiveField(structFields[i]).type();

while(nextFieldType instanceof Types.StructType && i < structFields.length) {

subField = caseSensitive ? subField.field(structFields[i]).type().asStructType() :
subField.caseInsensitiveField(structFields[i]).type().asStructType();

i++;

nextFieldType = caseSensitive ? subField.field(structFields[i]).type() :
subField.caseInsensitiveField(structFields[i]).type();
}

return subField.field(lastFieldInPath);

}

@xabriel xabriel Mar 11, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Now that we need to recurse, I wonder if it makes sense to reuse the indexes available in TypeUtils: https://github.com/apache/incubator-iceberg/blob/0c9c63140e838875dc8cc52a57be2c8f24ad9975/api/src/main/java/com/netflix/iceberg/types/TypeUtil.java#L78-L84 so that this code simplifies to:

Integer idx = indexByName.get(expressionFieldPath);
Types.NestedField field = indexById.get(idx);

That code doesn't check for non-struct parents; we'd have to perhaps create a custom Indexer. Also we would need to calculate the index lazily and perhaps cache it ( As in the Schema, see https://github.com/apache/incubator-iceberg/blob/master/api/src/main/java/com/netflix/iceberg/Schema.java#L59-L71 )

@prodeezy prodeezy Mar 13, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Doesn't look like TypeUtils.indexByName & TypeUtils.indexById are used when reading data. It's used by the Manifest reader to index on manifest file statistics data.

@prodeezy prodeezy Mar 14, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I guess you are suggesting we now start using this. i dunno what impact that would have though. do you see a major benefit to it? I can look into it.

@xabriel xabriel Mar 14, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I guess you are suggesting we now start using this.

Right.

do you see a major benefit to it?

Code reuse. But it would only make sense if findNestedField() is called many times. The first call to findNestedField() would build the indexes and consult them. Then any subsequent call to findNestedField() would just consult the cached Maps.

@prodeezy prodeezy Mar 18, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Right. seems like it's being called many times but by different expression evaluators. A cache to lookup field name to Types.NestedField would help. Good call. will add one in.

@prodeezy prodeezy Mar 26, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@xabriel I'm now creating an index that keeps ids to accessors and then use accessors to reach type and fields. access to this index is lazy as well.

}
12 changes: 0 additions & 12 deletions parquet/src/main/java/com/netflix/iceberg/parquet/ParquetDictionaryRowGroupFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,8 +161,6 @@ public <T> Boolean notNull(BoundReference<T> ref) {
@Override
public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand All @@ -185,8 +183,6 @@ public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand All @@ -209,8 +205,6 @@ public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand All @@ -233,8 +227,6 @@ public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand All @@ -257,8 +249,6 @@ public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand All @@ -273,8 +263,6 @@ public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Boolean hasNonDictPage = isFallback.get(id);
if (hasNonDictPage == null || hasNonDictPage) {
Expand Down
14 changes: 0 additions & 14 deletions parquet/src/main/java/com/netflix/iceberg/parquet/ParquetMetricsRowGroupFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,8 +131,6 @@ public <T> Boolean isNull(BoundReference<T> ref) {
// no need to check whether the field is required because binding evaluates that case
// if the column has no null values, the expression cannot match
Integer id = ref.fieldId();
Preconditions.checkNotNull(struct.field(id),
"Cannot filter by nested column: %s", schema.findField(id));

Long valueCount = valueCounts.get(id);
if (valueCount == null) {
Expand All @@ -154,8 +152,6 @@ public <T> Boolean notNull(BoundReference<T> ref) {
// no need to check whether the field is required because binding evaluates that case
// if the column has no non-null values, the expression cannot match
Integer id = ref.fieldId();
Preconditions.checkNotNull(struct.field(id),
"Cannot filter by nested column: %s", schema.findField(id));

// When filtering nested types notNull() is implicit filter passed even though complex
// filters aren't pushed down in Parquet. Leave all nested column type filters to be
Expand All @@ -182,8 +178,6 @@ public <T> Boolean notNull(BoundReference<T> ref) {
@Override
public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Long valueCount = valueCounts.get(id);
if (valueCount == null) {
Expand All @@ -210,8 +204,6 @@ public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Long valueCount = valueCounts.get(id);
if (valueCount == null) {
Expand All @@ -238,8 +230,6 @@ public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Long valueCount = valueCounts.get(id);
if (valueCount == null) {
Expand All @@ -266,8 +256,6 @@ public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

Long valueCount = valueCounts.get(id);
if (valueCount == null) {
Expand All @@ -294,8 +282,6 @@ public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
@Override
public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
Integer id = ref.fieldId();
Types.NestedField field = struct.field(id);
Preconditions.checkNotNull(field, "Cannot filter by nested column: %s", schema.findField(id));

// When filtering nested types notNull() is implicit filter passed even though complex
// filters aren't pushed down in Parquet. Leave all nested column type filters to be
Expand Down