Improve avro support #691
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
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
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 |
|---|---|---|
|
|
@@ -18,118 +18,167 @@ | |
| package com.spotify.ratatool.diffy | ||
|
|
||
| import com.spotify.scio.coders.Coder | ||
| import org.apache.avro.Schema | ||
| import org.apache.avro.generic.{GenericRecord, IndexedRecord} | ||
| import org.apache.avro.specific.SpecificData | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| /** | ||
| * Field level diff tool for Avro records. | ||
| * | ||
| * @param ignore | ||
| * specify set of fields to ignore during comparison. | ||
| * @param unordered | ||
| * a list of fields to be treated as unordered, i.e. sort before comparison. | ||
| * @param unorderedFieldKeys | ||
| * a map of record field names to fields names that can be keyed by when comparing nested repeated | ||
| * records. (currently not support in CLI) | ||
| */ | ||
| class AvroDiffy[T <: IndexedRecord: Coder]( | ||
| ignore: Set[String] = Set.empty, | ||
| unordered: Set[String] = Set.empty, | ||
| unorderedFieldKeys: Map[String, String] = Map() | ||
| ) extends Diffy[T](ignore, unordered, unorderedFieldKeys) { | ||
|
|
||
| override def apply(x: T, y: T): Seq[Delta] = (x, y) match { | ||
| case (null, null) => Seq.empty | ||
| case (_, null) => Seq(Delta("", Some(x), None, UnknownDelta)) | ||
| case (null, _) => Seq(Delta("", None, Some(y), UnknownDelta)) | ||
| case _ if x.getSchema != y.getSchema => Seq(Delta("", Some(x), Some(y), UnknownDelta)) | ||
| case _ => diff(x, y, x.getSchema, "") | ||
| } | ||
|
|
||
| private def isRecord(schema: Schema): Boolean = schema.getType match { | ||
| case Schema.Type.RECORD => true | ||
| case Schema.Type.UNION => schema.getTypes.asScala.map(_.getType).contains(Schema.Type.RECORD) | ||
| case _ => false | ||
| } | ||
|
|
||
| private def isNumericType(`type`: Schema.Type): Boolean = `type` match { | ||
| case Schema.Type.INT | Schema.Type.LONG | Schema.Type.FLOAT | Schema.Type.DOUBLE => true | ||
| case _ => false | ||
| } | ||
|
|
||
| private def numericValue(value: AnyRef): Double = value match { | ||
| case i: java.lang.Integer => i.toDouble | ||
|
There was a problem hiding this comment. Because numericDelta only supports double |
||
| case l: java.lang.Long => l.toDouble | ||
| case f: java.lang.Float => f.toDouble | ||
| case d: java.lang.Double => d | ||
| case _ => throw new IllegalArgumentException(s"Unsupported numeric type: ${value.getClass}") | ||
| } | ||
|
|
||
| private def diff(x: AnyRef, y: AnyRef, schema: Schema, field: String): Seq[Delta] = { | ||
benkonz marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| val deltas = schema.getType match { | ||
| case Schema.Type.UNION => | ||
| // union, must resolve to same type | ||
| val data = SpecificData.get() | ||
| val xTypeIndex = data.resolveUnion(schema, x) | ||
| val yTypeIndex = data.resolveUnion(schema, y) | ||
| if (xTypeIndex != yTypeIndex) { | ||
| // Use Option as x or y can be null | ||
| Seq(Delta(field, Option(x), Option(y), UnknownDelta)) | ||
| } else { | ||
| // same fields, refined schema | ||
| val fieldSchema = schema.getTypes.get(xTypeIndex) | ||
| diff(x, y, fieldSchema, field) | ||
| } | ||
|
|
||
| case Schema.Type.RECORD => | ||
| // record, compare all fields | ||
| val a = x.asInstanceOf[IndexedRecord] | ||
| val b = y.asInstanceOf[IndexedRecord] | ||
|
Comment on lines
+89
to
+90
There was a problem hiding this comment. I like this! Seems more resource-efficient. Is there any case in which this cast could not work, though, so to add in some catch statement? There was a problem hiding this comment. if schema type is a record, |
||
| for { | ||
| f <- schema.getFields.asScala.toSeq | ||
| pos = f.pos() | ||
| name = f.name() | ||
| fullName = if (field.isEmpty) name else field + "." + name | ||
| delta <- diff(a.get(pos), b.get(pos), f.schema(), fullName) | ||
| } yield delta | ||
|
|
||
| case Schema.Type.ARRAY | ||
| if unorderedFieldKeys.contains(field) && isRecord(schema.getElementType) => | ||
| // keyed array, compare like Map[String, Record] | ||
| val keyField = unorderedFieldKeys(field) | ||
| val as = | ||
| x.asInstanceOf[java.util.List[GenericRecord]].asScala.map(r => r.get(keyField) -> r).toMap | ||
| val bs = | ||
| y.asInstanceOf[java.util.List[GenericRecord]].asScala.map(r => r.get(keyField) -> r).toMap | ||
|
|
||
| for { | ||
| k <- (as.keySet ++ bs.keySet).toSeq | ||
| elementField = field + s"[$k]" | ||
| delta <- (as.get(k), bs.get(k)) match { | ||
| case (Some(a), Some(b)) => diff(a, b, schema.getElementType, field) | ||
| case (a, b) => Seq(Delta(field, a, b, UnknownDelta)) | ||
| } | ||
| } yield delta.copy(field = delta.field.replaceFirst(field, elementField)) | ||
|
|
||
| case Schema.Type.ARRAY => | ||
| // array, (un)ordered comparison | ||
| val xs = x.asInstanceOf[java.util.List[AnyRef]] | ||
| val ys = y.asInstanceOf[java.util.List[AnyRef]] | ||
| val (as, bs) = if (unordered.contains(field)) { | ||
| // ordered comparison | ||
| (sortList(xs).asScala, sortList(ys).asScala) | ||
| } else { | ||
| // unordered | ||
| (xs.asScala, ys.asScala) | ||
| } | ||
|
|
||
| val delta = if (as.size != bs.size) { | ||
| Some(UnknownDelta) | ||
| } else if (isNumericType(schema.getElementType.getType) && as != bs) { | ||
| Some(VectorDelta(vectorDelta(as.map(numericValue).toSeq, bs.map(numericValue).toSeq))) | ||
| } else if (as != bs) { | ||
| as.zip(bs) | ||
| .find { case (a, b) => | ||
| a != b && diff(a, b, schema.getElementType, field).nonEmpty | ||
| } | ||
| .map(_ => UnknownDelta) | ||
| } else { | ||
| None | ||
| } | ||
| delta.map(d => Delta(field, Some(x), Some(y), d)).toSeq | ||
|
|
||
| case Schema.Type.MAP => | ||
| // map, compare key set and values | ||
| val as = x.asInstanceOf[java.util.Map[CharSequence, AnyRef]].asScala.map { case (k, v) => | ||
| k.toString -> v | ||
| } | ||
| val bs = y.asInstanceOf[java.util.Map[CharSequence, AnyRef]].asScala.map { case (k, v) => | ||
| k.toString -> v | ||
| } | ||
|
|
||
| for { | ||
| k <- (as.keySet ++ bs.keySet).toSeq | ||
| elementField = field + s"[$k]" | ||
| delta <- (as.get(k), bs.get(k)) match { | ||
| case (Some(a), Some(b)) => diff(a, b, schema.getValueType, field) | ||
| case (a, b) => Seq(Delta(field, a, b, UnknownDelta)) | ||
| } | ||
| } yield delta.copy(field = delta.field.replaceFirst(field, elementField)) | ||
|
|
||
| case Schema.Type.STRING => | ||
| // string, convert to java String for equality check | ||
| val a = x.asInstanceOf[CharSequence].toString | ||
| val b = y.asInstanceOf[CharSequence].toString | ||
| val delta = if (a == b) None else Some(StringDelta(stringDelta(a, b))) | ||
| delta.map(d => Delta(field, Some(x), Some(y), d)).toSeq | ||
|
|
||
| case t if isNumericType(t) => | ||
| // numeric, convert to Double for equality check | ||
| val a = numericValue(x) | ||
| val b = numericValue(y) | ||
| val delta = if (a == b) None else Some(NumericDelta(numericDelta(a, b))) | ||
| delta.map(d => Delta(field, Some(x), Some(y), d)).toSeq | ||
|
|
||
| case _ => | ||
| // other case rely on object equality | ||
| val delta = if (x == y) None else Some(UnknownDelta) | ||
| delta.map(d => Delta(field, Some(x), Some(y), d)).toSeq | ||
| } | ||
|
|
||
| deltas.filterNot(d => ignore.contains(d.field)) | ||
| } | ||
| } | ||
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.
Were there any breaking changes on this implementation, or is there any other reason why we're moving away from avro schema validation? I'm not very familiar with how strict is the avro definition of "compatible schemas", but at first glance it seems like we're loosing some flexibility and/or some level of detail with the new validations. This is not my area of expertise, though, so your recommendations are more than welcome!
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.
In terms of diff, avro schemas must be strictly equal. This is checked line 62 in the new version.
Compatible schema are used on read time to adapt stored data to the desired read schema. Once in memory, we should not compare data constructed with different schema, even if compatible.
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.
Can you expand on what you mean? I'm a bit confused reading this thread
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.
avro record with different models must not be compared.
schema compatibility is relevant when reading, making sure the
writerSchemaand thereaderSchemaare compatible. Once read, the records strictly folow thereaderSchemawhere field index matters. Strict schema equality must be ensured before comparing content.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.
Ok, trying to confirm my understanding here since I'm still a bit confused.
Dataset A is updated to add new nullable field
xand becomes Dataset A'.We go to diff these two datasets.
Are you saying that this field comparison will end up in one of the above null cases prior to the schema comparison?
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.
I would go one step further and say it should necessarily be the A' schema, even if it's nullable/has default, and cases where a field is missing should fail. Semantically, it's still a difference between the two datasets. IIRC this is the current functionality. It's unclear to me if/where this behaviour is retained
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.
The BigDiffy API of this lib is not file aware. It only works in terms of in-memory records and can't make any assumption on writer schema.
It is up to the users creating the
SCollectionto read using the correct schemaThere 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.
Even for the diffAvro API, the reader schema used is the one from the generated class.
It is totally possible that underlying files are using a different schema,
ratatool-diffywill miss those.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.
cc @RustedBones, AFAIK, BigDiffy is file-aware when run through the CLI, which invokes BigDiffy#run:
ratatool/ratatool-diffy/src/main/scala/com/spotify/ratatool/diffy/BigDiffy.scala
Lines 751 to 760 in 51ec798
but even then, it looks like it selects the schema associated with the RHS and uses that for both resulting SCollections. So maybe we could add schema validation there (ensure that RHS schema is equal to, or a superset of, the LHS schema)?
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.
Added an extra check that will prefer backward compatible reader schema.
When schemas are different, but both forward and backward compatible, will print a warning