Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu

@Override
public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
if (fieldResult == null) {
if (field.type().isStructType() || field.type().isPrimitiveType()) {
fieldIds.add(field.fieldId());
}
return fieldIds;
Expand Down
4 changes: 2 additions & 2 deletions api/src/main/java/org/apache/iceberg/types/TypeUtil.java
Original file line number Diff line number Diff line change
Expand Up @@ -132,13 +132,13 @@ private static Set<Integer> getIdsInternal(Type type) {
public static Types.StructType selectNot(Types.StructType struct, Set<Integer> fieldIds) {
Set<Integer> projectedIds = getIdsInternal(struct);
projectedIds.removeAll(fieldIds);
return select(struct, projectedIds);
return project(struct, projectedIds);

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

One issue here is selectNot doest not actually deselect children when a parent ID is not selected. Previously this is because getProjectedIDs (behind getIdsInternal) would not return parent struct ids, so removing it from the set of projectedIds would not do anything.

Now It will not work because removing a parentID still leaves all child IDs. We could fix this but it would be a change in behavior from the previous code.

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 think I agree with the decision to not change the behavior of this method, even though the opposite of "select" behavior would be to fully remove a struct when its ID is passed in fieldIds.

But I don't think that project is quite correct either. Consider the example schema 1: id bigint, 2: location struct<3: lat double, 4: long double>. Previously, selectNot(t, set(3, 4)) would produce 1: id bigint and omit the location entirely. Using project with the updated GetProjectedIds, the projected ID set will be {1, 2, 3, 4} and not {1, 3, 4}. That would result in the same call producing 1: id bigint, 2: location struct<>, which introduces a new bug because now there is an unexpected extra field.

To clean this up, I think we need a version of GetProjectedIds that doesn't select structs and uses the old behavior.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

That seems like the right behavior to me? Shouldn't you be required to explicitly omit the parent if you don't want the that element? Otherwise there would be no way to "selectNot" and only get back the empty struct.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Wrote up these test cases, i'll run the full test suite to make sure this works with our other usages

    Schema schema = new Schema(
        Lists.newArrayList(
            required(1, "id", Types.LongType.get()),
            required(2, "location", Types.StructType.of(
                required(3, "lat", Types.DoubleType.get()),
                required(4, "long", Types.DoubleType.get())
            ))));

    Schema expectedNoPrimitive = new Schema(
        Lists.newArrayList(
            required(2, "location", Types.StructType.of(
                required(3, "lat", Types.DoubleType.get()),
                required(4, "long", Types.DoubleType.get())
            ))));

    Schema actualNoPrimitve = TypeUtil.selectNot(schema, Sets.newHashSet(1));
    Assert.assertEquals(expectedNoPrimitive.asStruct(), actualNoPrimitve.asStruct());

    // Expected legacy behavior is to completely remove structs if their elements are removed
    Schema expectedNoStructElements = new Schema(required(1, "id", Types.LongType.get()));
    Schema actualNoStructElements = TypeUtil.selectNot(schema, Sets.newHashSet(3, 4));
    Assert.assertEquals(expectedNoStructElements.asStruct(), actualNoStructElements.asStruct());

    // Expected legacy behavior is to ignore selectNot on struct elements.
    Schema actualNoStruct = TypeUtil.selectNot(schema, Sets.newHashSet(2));
    Assert.assertEquals(schema.asStruct(), actualNoStruct.asStruct());
    ```

}

public static Schema selectNot(Schema schema, Set<Integer> fieldIds) {
Set<Integer> projectedIds = getIdsInternal(schema.asStruct());
projectedIds.removeAll(fieldIds);
return select(schema, projectedIds);
return project(schema, projectedIds);
}

public static Schema join(Schema left, Schema right) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public class StructProjection implements StructLike {
*/
public static StructProjection create(Schema schema, Set<Integer> ids) {
StructType structType = schema.asStruct();
return new StructProjection(structType, TypeUtil.select(structType, ids));
return new StructProjection(structType, TypeUtil.project(structType, ids));

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.

Looks like there aren't any uses of this call, which is good. I agree that we probably want this to use project instead of select.

}

/**
Expand Down
49 changes: 13 additions & 36 deletions api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

package org.apache.iceberg.types;

import java.util.Set;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.Schema;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
Expand Down Expand Up @@ -323,48 +324,24 @@ public void testProjectMap() {
}

@Test
public void testProjectList() {
public void testGetProjectedIds() {
Schema schema = new Schema(
Lists.newArrayList(
required(10, "a", Types.IntegerType.get()),
required(11, "A", Types.IntegerType.get()),
required(12, "list", Types.ListType.ofRequired(13,
Types.StructType.of(
optional(20, "foo", Types.IntegerType.get()),
required(21, "subList", Types.ListType.ofRequired(14,
Types.StructType.of(
required(15, "x", Types.IntegerType.get()),
required(16, "y", Types.IntegerType.get()),
required(17, "z", Types.IntegerType.get())))))))));


AssertHelpers.assertThrows("Cannot explicitly project List",
IllegalArgumentException.class,
() -> TypeUtil.project(schema, Sets.newHashSet(12))
);

AssertHelpers.assertThrows("Cannot explicitly project List",
IllegalArgumentException.class,
() -> TypeUtil.project(schema, Sets.newHashSet(21))
);
required(35, "emptyStruct", Types.StructType.of()),
required(12, "someStruct", Types.StructType.of(
required(13, "b", Types.IntegerType.get()),
required(14, "B", Types.IntegerType.get()),
required(15, "anotherStruct", Types.StructType.of(
required(16, "c", Types.IntegerType.get()),
required(17, "C", Types.IntegerType.get()))
)))));

Schema expectedDepthOne = new Schema(
Lists.newArrayList(
required(12, "list", Types.ListType.ofRequired(13,
Types.StructType.of()))));
Schema actualDepthOne = TypeUtil.project(schema, Sets.newHashSet(13));
Assert.assertEquals(expectedDepthOne.asStruct(), actualDepthOne.asStruct());
Set<Integer> expectedIds = Sets.newHashSet(10, 11, 35, 12, 13, 14, 15, 16, 17);
Set<Integer> actualIds = TypeUtil.getProjectedIds(schema);

Schema expectedDepthTwo = new Schema(
Lists.newArrayList(
required(10, "a", Types.IntegerType.get()),
required(12, "list", Types.ListType.ofRequired(13,
Types.StructType.of(
optional(20, "foo", Types.IntegerType.get()),
required(21, "subList", Types.ListType.ofRequired(14,
Types.StructType.of())))))));
Schema actualDepthTwo = TypeUtil.project(schema, Sets.newHashSet(10, 13, 20, 14));
Assert.assertEquals(expectedDepthTwo.asStruct(), actualDepthTwo.asStruct());
Assert.assertEquals(expectedIds, actualIds);
}

@Test
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/java/org/apache/iceberg/BaseTableScan.java
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,7 @@ private Schema lazyColumnProjection() {
}
requiredFieldIds.addAll(selectedIds);

return TypeUtil.select(schema, requiredFieldIds);
return TypeUtil.project(schema, requiredFieldIds);

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 agree with this because it is the opposite of GetProjectedIds used above.


} else if (context.projectedSchema() != null) {
return context.projectedSchema();
Expand Down
9 changes: 7 additions & 2 deletions core/src/main/java/org/apache/iceberg/avro/PruneColumns.java
Original file line number Diff line number Diff line change
Expand Up @@ -81,13 +81,18 @@ public Schema record(Schema record, List<String> names, List<Schema> fields) {

Schema fieldSchema = fields.get(field.pos());
// All primitives are selected by selecting the field, but map and list
// types can be selected by projecting the keys, values, or elements.
// types can be selected by projecting the keys, values, or elements. Empty
// Structs can be selected by selecting the record itself instead of it's children.
// This creates two conditions where the field should be selected: if the
// id is selected or if the result of the field is non-null. The only
// case where the converted field is non-null is when a map or list is
// selected by lower IDs.
if (selectedIds.contains(fieldId)) {
filteredFields.add(copyField(field, field.schema(), fieldId));
if (fieldSchema != null) {
filteredFields.add(copyField(field, fieldSchema, fieldId));

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'm not sure that I understand the reason for this change. Is this implementing the same change as the previous PR, but in the Avro PruneColumns?

It looks like if a struct field is selected and a sub-field is selected, then the selection for the struct isn't a full selection. But if a sub-field is not selected then the selection for the struct is a full selection. That doesn't make sense to me.

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.

As I'm thinking about this more, I think that the behavior in this class should always match project. I doubt there's a case where we want select behavior, right? In that case, shouldn't the else case check whether the type is a record and create an empty record?

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 think we need to set hasChange in the cases where we don't return field.schema() for the field, right?

@RussellSpitzer RussellSpitzer Sep 20, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I think we are actually fine here unless every field is selected because the logic for has change is a bit confusing.

You either

  1. Have a change (Make a new record using the filtered fields) ( Return Changed Records)
  2. Have no change and filtered fields size is the same as the original number of fields ( Return Original Record)
  3. Have no change and filtered field size is not empty (Make a new record using the filtered fields) (Return changed record)

Currently we have tests hitting 1 and 3 but not 2 :/
I'll add the "hasChange" flag

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.

This looks good now.

} else {
filteredFields.add(copyField(field, field.schema(), fieldId));
}
} else if (fieldSchema != null) {
hasChange = true;
filteredFields.add(copyField(field, fieldSchema, fieldId));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public void testDeleteFields() {
Schema del = new SchemaUpdate(SCHEMA, 19).deleteColumn(name).apply();

Assert.assertEquals("Should match projection with '" + name + "' removed",
TypeUtil.select(SCHEMA, selected).asStruct(), del.asStruct());
TypeUtil.project(SCHEMA, selected).asStruct(), del.asStruct());
Comment thread
rdblue marked this conversation as resolved.
}
}

Expand Down
186 changes: 183 additions & 3 deletions core/src/test/java/org/apache/iceberg/avro/TestReadProjection.java
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,11 @@
import org.junit.rules.TemporaryFolder;

public abstract class TestReadProjection {

protected abstract Record writeAndRead(String desc,
Schema writeSchema,
Schema readSchema,
Record record) throws IOException;
Schema writeSchema,
Schema readSchema,
Record record) throws IOException;
Comment thread
rdblue marked this conversation as resolved.
Outdated

@Rule
public TemporaryFolder temp = new TemporaryFolder();
Expand Down Expand Up @@ -526,4 +527,183 @@ public void testListOfStructsProjection() throws IOException {
AssertHelpers.assertEmptyAvroField(projectedP2, "y");
Assert.assertNull("Should project null z", projectedP2.get("z"));
}

@Test
public void testEmptyStructProjection() throws Exception {
Schema writeSchema = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.optional(3, "location", Types.StructType.of(
Types.NestedField.required(1, "lat", Types.FloatType.get()),
Types.NestedField.required(2, "long", Types.FloatType.get())
))
);

Record record = new Record(AvroSchemaUtil.convert(writeSchema, "table"));
record.put("id", 34L);
Record location = new Record(
AvroSchemaUtil.fromOption(record.getSchema().getField("location").schema()));
location.put("lat", 52.995143f);
location.put("long", -1.539054f);
record.put("location", location);

Schema emptyStruct = new Schema(
Types.NestedField.required(3, "location", Types.StructType.of())
);

Record projected = writeAndRead("empty_proj", writeSchema, emptyStruct, record);
AssertHelpers.assertEmptyAvroField(projected, "id");
Record result = (Record) projected.get("location");

Assert.assertEquals("location should be in the 0th position", result, projected.get(0));
Assert.assertNotNull("Should contain an empty record", result);
AssertHelpers.assertEmptyAvroField(result, "lat");
AssertHelpers.assertEmptyAvroField(result, "long");
}

@Test
public void testEmptyStructRequiredProjection() throws Exception {

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.

Isn't this identical to the test case above?

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.

Nevermind, I see that the write schema has the struct as optional.

Schema writeSchema = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.required(3, "location", Types.StructType.of(
Types.NestedField.required(1, "lat", Types.FloatType.get()),
Types.NestedField.required(2, "long", Types.FloatType.get())
))
);

Record record = new Record(AvroSchemaUtil.convert(writeSchema, "table"));
record.put("id", 34L);
Record location = new Record(record.getSchema().getField("location").schema());
location.put("lat", 52.995143f);
location.put("long", -1.539054f);

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.

Odd location to choose.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

It was already in the test suite and came in the Netflix original commit :) So you'll have to ask whoever wrote the first version.

https://github.com/apache/iceberg/blame/master/core/src/test/java/org/apache/iceberg/avro/TestReadProjection.java#L205-L206

record.put("location", location);

Schema emptyStruct = new Schema(
Types.NestedField.required(3, "location", Types.StructType.of())
);

Record projected = writeAndRead("empty_req_proj", writeSchema, emptyStruct, record);
AssertHelpers.assertEmptyAvroField(projected, "id");
Record result = (Record) projected.get("location");
Assert.assertEquals("location should be in the 0th position", result, projected.get(0));
Assert.assertNotNull("Should contain an empty record", result);
AssertHelpers.assertEmptyAvroField(result, "lat");
AssertHelpers.assertEmptyAvroField(result, "long");
}

@Test
public void testRequiredEmptyStructInRequiredStruct() throws Exception {
Schema writeSchema = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.required(3, "location", Types.StructType.of(
Types.NestedField.required(1, "lat", Types.FloatType.get()),
Types.NestedField.required(2, "long", Types.FloatType.get()),
Types.NestedField.required(4, "empty", Types.StructType.of())
))
);

Record record = new Record(AvroSchemaUtil.convert(writeSchema, "table"));
record.put("id", 34L);
Record location = new Record(record.getSchema().getField("location").schema());
location.put("lat", 52.995143f);
location.put("long", -1.539054f);
record.put("location", location);

Schema emptyStruct = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.required(3, "location", Types.StructType.of(
Types.NestedField.required(4, "empty", Types.StructType.of())
))
);

Record projected = writeAndRead("req_empty_req_proj", writeSchema, emptyStruct, record);
Assert.assertEquals("Should project id", 34L, projected.get("id"));
Record result = (Record) projected.get("location");
Assert.assertEquals("location should be in the 1st position", result, projected.get(1));
Assert.assertNotNull("Should contain an empty record", result);
AssertHelpers.assertEmptyAvroField(result, "lat");
AssertHelpers.assertEmptyAvroField(result, "long");
Assert.assertNotNull("Should project empty", result.getSchema().getField("empty"));
Assert.assertNotNull("Empty should not be null", result.get("empty"));
Assert.assertEquals("Empty should be empty", 0,
((Record) result.get("empty")).getSchema().getFields().size());
}

@Test
public void testEmptyNestedStructProjection() throws Exception {
Schema writeSchema = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.optional(3, "outer", Types.StructType.of(
Types.NestedField.required(1, "lat", Types.FloatType.get()),
Types.NestedField.optional(2, "inner", Types.StructType.of(
Types.NestedField.required(5, "lon", Types.FloatType.get())
)
)
))
);

Record record = new Record(AvroSchemaUtil.convert(writeSchema, "table"));
record.put("id", 34L);
Record outer = new Record(
AvroSchemaUtil.fromOption(record.getSchema().getField("outer").schema()));
Record inner = new Record(AvroSchemaUtil.fromOption(outer.getSchema().getField("inner").schema()));
inner.put("lon", 32.14f);
outer.put("lat", 52.995143f);
outer.put("inner", inner);
record.put("outer", outer);

Schema emptyStruct = new Schema(
Types.NestedField.required(3, "outer", Types.StructType.of(
Types.NestedField.required(2, "inner", Types.StructType.of())
)));

Record projected = writeAndRead("nested_empty_proj", writeSchema, emptyStruct, record);
AssertHelpers.assertEmptyAvroField(projected, "id");
Record outerResult = (Record) projected.get("outer");
Assert.assertEquals("Outer should be in the 0th position", outerResult, projected.get(0));
Assert.assertNotNull("Should contain the outer record", outerResult);
AssertHelpers.assertEmptyAvroField(outerResult, "lat");
Record innerResult = (Record) outerResult.get("inner");
Assert.assertEquals("Inner should be in the 0th position", innerResult, outerResult.get(0));
Assert.assertNotNull("Should contain the inner record", innerResult);
AssertHelpers.assertEmptyAvroField(innerResult, "lon");
}

@Test
public void testEmptyNestedStructRequiredProjection() throws Exception {
Schema writeSchema = new Schema(
Types.NestedField.required(0, "id", Types.LongType.get()),
Types.NestedField.required(3, "outer", Types.StructType.of(
Types.NestedField.required(1, "lat", Types.FloatType.get()),
Types.NestedField.required(2, "inner", Types.StructType.of(
Types.NestedField.required(5, "lon", Types.FloatType.get())
)
)
))
);

Record record = new Record(AvroSchemaUtil.convert(writeSchema, "table"));
record.put("id", 34L);
Record outer = new Record(record.getSchema().getField("outer").schema());
Record inner = new Record(outer.getSchema().getField("inner").schema());
inner.put("lon", 32.14f);
outer.put("lat", 52.995143f);
outer.put("inner", inner);
record.put("outer", outer);

Schema emptyStruct = new Schema(
Types.NestedField.required(3, "outer", Types.StructType.of(
Types.NestedField.required(2, "inner", Types.StructType.of())
)));

Record projected = writeAndRead("nested_empty_req_proj", writeSchema, emptyStruct, record);
AssertHelpers.assertEmptyAvroField(projected, "id");
Record outerResult = (Record) projected.get("outer");
Assert.assertEquals("Outer should be in the 0th position", outerResult, projected.get(0));
Assert.assertNotNull("Should contain the outer record", outerResult);
AssertHelpers.assertEmptyAvroField(outerResult, "lat");
Record innerResult = (Record) outerResult.get("inner");
Assert.assertEquals("Inner should be in the 0th position", innerResult, outerResult.get(0));
Assert.assertNotNull("Should contain the inner record", innerResult);
AssertHelpers.assertEmptyAvroField(innerResult, "lon");
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,12 @@ public Type message(MessageType message, List<Type> fields) {
Type field = fields.get(i);
Integer fieldId = getId(originalField);
if (fieldId != null && selectedIds.contains(fieldId)) {
builder.addField(originalField);
if (field != null) {
hasChange = true;
builder.addField(field);
} else {
builder.addField(originalField);

@RussellSpitzer RussellSpitzer Sep 16, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Should I do the empty message only thing here as well? where we copy the struct to be empty?

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 do you mean?

}
fieldCount += 1;
} else if (field != null) {
builder.addField(field);
Expand Down
Loading