Support reading Avro complex union types #65

funcheetah · 2021-04-27T21:02:41Z

This PR supports reading Avro complex union types as Iceberg Struct type via SparkAvroReader.
For example,
[ "int", "string" ] => required struct ( 0 tag_0: optional int, 1 tag_1: optional string)
[ "null", "int", "string" ] => optional struct ( 0 tag_0: optional int, 1 tag_1: optional string)

Testing done:

unit test
integration test via spark reading table with non-option union

shenodaguirguis · 2021-05-04T20:20:43Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

    return false;
  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {


isNonOptionalUnionOptional is confusing. may be isNonOptionalUnion?

isNonOptionalUnionOptional is checking whether non option union is optional. It does more than isNonOptionalUnion . For example, [“int”, “string”] is non option union but it is required. [“null”, “int”, “string”] is non option union and it is optional.

I get the optional vs required part, but what is the non-option union part?
can we rename to isUnionOptional?

Option union only refers to [null, type]. While [int, string], [int], [null, int, string] etc are all non-option unions. What this method checks is whether a non-option union is optional. I think the original name is more accurate.

I see. Option is an Iceberg (bad) term == Optional with 2 types only.
Would reversing logic and renaming to isNonOptionUniontRequired make more sense? I will leave it to your preference. But if you keep it as isNonOptionUnionOptional, please clarify in a java-doc the difference between optional and option/non-option

I think we should rename this method. How about isOptionalComplexUnion?

Also, since the input is an Avro schema, should not we use Avro terminology? Also something along the lines of isNullableComplexUnion should work probably.

Thanks for the suggestion. Renamed it to isOptionalComplexUnion. Also added java doc to explain the semantics to make ti clearer.

shenodaguirguis · 2021-05-04T20:40:35Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java


      default:
-        return visitor.primitive(iType != null ? iType.asPrimitiveType() : null, schema);
+        if (iType == null || iType.isStructType()) {


doesn't iType.structType() map to case RECORD: in line 37?

I'm confused by adding calling isStructType here, could you clarify the logic here?

When visitUnion, each element of UNION will be visited along with iType as StructType. Thus, we need to handle iType properly. Discussed offline, we decide to move the StructType handling logic to visitUnion so that no change need to be made here.

shenodaguirguis · 2021-05-04T20:41:13Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java

      default:
-        return visitor.primitive(iType != null ? iType.asPrimitiveType() : null, schema);
+        if (iType == null || iType.isStructType()) {
+          return visitor.primitive(null, schema);


why do we need to pass null as opposed to iType.asStructType() in case of iType.isStructType() ?

When visitUnion, each element of UNION will be visited along with iType as StructType. Thus, we need to handle iType properly. Discussed offline, we decide to move the StructType handling logic to visitUnion so that no change need to be made here.

shenodaguirguis · 2021-05-04T22:04:48Z

core/src/main/java/org/apache/iceberg/avro/SchemaToType.java

+        }
+      }
+
+      return Types.StructType.of(newFields);


not sure I follow the reasoning why struct is the appropriate way to represent non-optional union. Should we introduce a new Types.unionType (that extends struct may be)? Because using struct per se (with optional fields) is somewhat confusing

Discussed offline with @shenodaguirguis . The reasoning is documented in design doc. The non-optional union can be represented with struct type as the following example
[ “null”, “int”, “string” ] => optional struct ( 1 tag_0: optional int, 2 tag_1: optional string)

shenodaguirguis · 2021-05-04T22:18:08Z

core/src/test/java/org/apache/iceberg/avro/TestAvroNonOptionalUnion.java

+import org.junit.Test;
+
+
+public class TestAvroNonOptionalUnion {


can we add tests for invalid cases as well?

Add more tests including non-option union.

shenodaguirguis · 2021-05-04T22:28:31Z

spark/src/main/java/org/apache/iceberg/spark/data/SparkValueReaders.java

+      Object value = this.readers[index].read(decoder, reuse);
+
+      struct.update(index, value);
+      for (int i = 0; i < readers.length && i != index; i += 1) {


should we do this for all i before we set the value at index? This loop is setting null all fields before the value, but not after...

Thanks for catching this. Change to explicitly set all fields to null before setting value at index.

rzhang10

Just want to add a general comment here, so I see this is to support reading unions in avro format data in spark, are we also going to support reading union in orc format in spark? If that's the case, I feel the scope will expand since we are merging hive and avro schema literal as the iceberg schema for orc format, and if the hive schema has a union it will throw and exception as unsupported.

rzhang10 · 2021-05-14T20:47:19Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {
+    if (schema.getType() == UNION && schema.getTypes().size() != 2) {


Could there be a corner case where the schema is a union that only contains null, like this:

"type": [ "null" ]

If so, it seems this will make the related code in SchemaToType.java return an empty struct when converting an union.

good catch @rzhang10 ! While it is meaningless to have a single-type in a union, the avro Schema.UnionSchema constructor does not perform any checks on the number of types, so we definitely need to guard against such cases (throw if size < 2 ?).

Yeah, it can happen. For example, [null] and [int] will pass the check. However, those are valid Avro unions and Avro does not prevent users from creating those schemas. In addition, our reader will works with those single unions. Maybe it is better to support those than guard it? We can let the user to decide whether it is useful or not.

Should those be converted to structs or simple types?

I'd prefer to convert it to struct for consistency as it is valid union in avro.

rzhang10 · 2021-05-14T20:51:14Z

core/src/main/java/org/apache/iceberg/avro/SchemaToType.java

+      } else {
+        return options.get(0);
+      }
    } else {


In this else branch, if it's the corner case (one null union) I mentioned above, it will construct an empty struct, which will prob cause an exception later.

Yes, it will create an empty struct. However, it is a valid Avro schema and our reader can read it without issue. Should we support it instead of guard against it?

thanks for validating this Wenye. I am fine with not guarding against it, given that avro schema also does not guard against this corner case.

core/src/main/java/org/apache/iceberg/avro/SchemaToType.java

rzhang10 · 2021-05-14T21:44:58Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java


      default:
-        return visitor.primitive(iType != null ? iType.asPrimitiveType() : null, schema);
+        if (iType == null || iType.isStructType()) {


I'm confused by adding calling isStructType here, could you clarify the logic here?

shenodaguirguis

Thanks for the changes @funcheetah !
LGTM in general. Other than the nits, the only - small - open issue I see is to guard against the wrong corner case of union type with < 2 types which @rzhang10 pointed out

shenodaguirguis · 2021-06-03T16:44:34Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

    return false;
  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {


I get the optional vs required part, but what is the non-option union part?
can we rename to isUnionOptional?

shenodaguirguis · 2021-06-03T17:00:18Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {
+    if (schema.getType() == UNION && schema.getTypes().size() != 2) {


good catch @rzhang10 ! While it is meaningless to have a single-type in a union, the avro Schema.UnionSchema constructor does not perform any checks on the number of types, so we definitely need to guard against such cases (throw if size < 2 ?).

shenodaguirguis · 2021-06-03T17:09:23Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java

+    int index = 0;
    for (Schema branch : types) {
      if (branch.getType() == Schema.Type.NULL) {
        options.add(visit((Type) null, branch, visitor));
      } else {
-        options.add(visit(type, branch, visitor));
+        if (AvroSchemaUtil.isOptionSchema(union)) {
+          options.add(visit(type, branch, visitor));
+        } else {
+          options.add(visit(type.asStructType().fields().get(index).type(), branch, visitor));
+        }
+        index++;


nit: we now maintain index outside the loop, and check if union is option (line 88) every time in the loop. May be use one loop for both branches and fields, and re-org to hoist up the union check (if that is feasible)?

How about re-assigning the type before the loop?

if (AvroSchemaUtil.isOptionSchema(union)) branches = type; else branches = type.asStructType();

shenodaguirguis · 2021-06-03T17:15:10Z

core/src/main/java/org/apache/iceberg/avro/SchemaToType.java

+      } else {
+        return options.get(0);
+      }
    } else {


wmoustafa · 2021-06-03T22:38:37Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {
+    if (schema.getType() == UNION && schema.getTypes().size() != 2) {


Should those be converted to structs or simple types?

wmoustafa · 2021-06-04T03:30:32Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

    return false;
  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {


I think we should rename this method. How about isOptionalComplexUnion?

wmoustafa · 2021-06-04T03:33:39Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

    return false;
  }

+  public static boolean isNonOptionalUnionOptional(Schema schema) {


Also, since the input is an Avro schema, should not we use Avro terminology? Also something along the lines of isNullableComplexUnion should work probably.

wmoustafa · 2021-06-04T03:48:40Z

core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java

+    int index = 0;
    for (Schema branch : types) {
      if (branch.getType() == Schema.Type.NULL) {
        options.add(visit((Type) null, branch, visitor));
      } else {
-        options.add(visit(type, branch, visitor));
+        if (AvroSchemaUtil.isOptionSchema(union)) {
+          options.add(visit(type, branch, visitor));
+        } else {
+          options.add(visit(type.asStructType().fields().get(index).type(), branch, visitor));
+        }
+        index++;


How about re-assigning the type before the loop?

if (AvroSchemaUtil.isOptionSchema(union)) branches = type; else branches = type.asStructType();

wmoustafa · 2021-06-04T03:51:01Z

core/src/main/java/org/apache/iceberg/avro/PruneColumns.java

+
+      return null;
+    } else {
+      // case non option union


/// Complex union case

Thanks. Modified.

wmoustafa · 2021-06-04T05:39:24Z

core/src/test/java/org/apache/iceberg/avro/TestAvroNonOptionalUnion.java

+  }
+
+  @Test
+  public void testNonOptionUnionNullable() {


testOptionalComplexUnion

wmoustafa · 2021-06-04T05:39:52Z

core/src/test/java/org/apache/iceberg/avro/TestAvroNonOptionalUnion.java

+  }
+
+  @Test
+  public void testSingleOptionUnion() {


testSingleComponentUnion

wmoustafa · 2021-06-04T05:42:41Z

spark/src/test/java/org/apache/iceberg/spark/data/TestSparkAvroUnions.java

+  public TemporaryFolder temp = new TemporaryFolder();
+
+  @Test
+  public void writeAndValidateNonOptionUnionNonNullable() throws IOException {


RequiredComplexUnion

wmoustafa · 2021-06-04T05:43:18Z

spark/src/test/java/org/apache/iceberg/spark/data/TestSparkAvroUnions.java

+  }
+
+  @Test
+  public void writeAndValidateNonOptionUnionNullable() throws IOException {


OptionalComplexUnion

wmoustafa · 2021-06-04T05:43:37Z

spark/src/test/java/org/apache/iceberg/spark/data/TestSparkAvroUnions.java

+  }
+
+  @Test
+  public void writeAndValidateSingleOptionUnion() throws IOException {


SingleComponent

funcheetah requested a review from shardulm94 April 27, 2021 21:08

shenodaguirguis reviewed May 4, 2021

View reviewed changes

rzhang10 reviewed May 14, 2021

View reviewed changes

Wenye Zhang added 6 commits June 1, 2021 15:18

support read Avro non-option union type

d45c91c

fix checkstyle issue

82f06fe

fix failing tests

bbdacef

add tests

d3bf9b7

add more tests

0e06bf6

address comments

790874c

funcheetah force-pushed the support-union-type branch from 0536543 to 790874c Compare June 1, 2021 22:24

Wenye Zhang added 2 commits June 2, 2021 13:24

fix failing tests

a206064

fix checkstyle

dd07a2a

shenodaguirguis reviewed Jun 3, 2021

View reviewed changes

shenodaguirguis approved these changes Jun 3, 2021

View reviewed changes

wmoustafa reviewed Jun 4, 2021

View reviewed changes

Wenye Zhang added 2 commits June 4, 2021 12:28

address comments

67150a4

fix checkstyle

cc7efc9

wmoustafa changed the title ~~support read Avro non-option union type~~ Support reading Avro complex union types Jun 4, 2021

wmoustafa approved these changes Jun 4, 2021

View reviewed changes

funcheetah merged commit 5e724ae into linkedin:master Jun 4, 2021

This was referenced Jun 4, 2021

Support reading Avro complex union types #73

Merged

Support reading ORC complex union types #74

Merged

		import org.junit.Test;


		public class TestAvroNonOptionalUnion {

Support reading Avro complex union types #65

Support reading Avro complex union types #65

Uh oh!

Conversation

funcheetah commented Apr 27, 2021 • edited by wmoustafa Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shenodaguirguis May 4, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rzhang10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shenodaguirguis left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

funcheetah commented Apr 27, 2021 •

edited by wmoustafa

Loading

shenodaguirguis May 4, 2021 •

edited

Loading