CDAP-16708 add API for autojoiner #12185
Conversation
albertshau
force-pushed
the
feature_release/CDAP-16708-new-joiner-api
branch
from
e5f8bb5
to
ad1ec9a
Compare
| return keys; | ||
| } | ||
|
|
||
| public boolean shouldDropNullKeys() { |
There was a problem hiding this comment.
For boolean field, use isDropNullKeys().
| throw new InvalidJoinConditionException(String.format( | ||
| "Join key for stage '%s' is invalid. Field '%s' does not exist in the stage.", | ||
| joinStageName, keysCopy.iterator().next())); | ||
| } else if (keysCopy.size() > 1) { |
There was a problem hiding this comment.
No need to have else after a throw inside the if block.
There was a problem hiding this comment.
this is an else-if, since it's also possible the set is empty, in which case it's valid. I can restructure this a little to make it more clear
There was a problem hiding this comment.
I mean it's cleaner to write:
if (keyCopy.size() == 1) {
throw
}
if (keyCopy.size() > 1) {
throw
}
| public class JoinStage { | ||
| private final String stageName; | ||
| private final Schema schema; | ||
| private final boolean isRequired; |
There was a problem hiding this comment.
Just call it required (so that the getter is isRequired).
| private final String stageName; | ||
| private final Schema schema; | ||
| private final boolean isRequired; | ||
| private final boolean shouldBroadcast; |
| * The name of a field and an optional alias to rename it to. | ||
| */ | ||
| @Beta | ||
| public class Field { |
There was a problem hiding this comment.
Prefer to call this JoinField.
| return this; | ||
| } | ||
|
|
||
| public Builder setShouldDropNullKeys(boolean dropNullKeys) { |
There was a problem hiding this comment.
Better match the method name with the variable. setDropNullKeys.
| // this means there are stages with different number of fields. | ||
| // it's the equivalent of trying to join on A.id = B.id and A.name = [null] | ||
| if (numFieldsToStages.size() > 1) { | ||
| Iterator<Integer> keyIter = numFieldsToStages.keySet().iterator(); |
There was a problem hiding this comment.
I think it's better to iterate through the numFioeldsToStages to use all stages to form the error message. E.g. if the map is like this:
{
1: ["s1"],
2: ["s2"],
3: ["s3"]
}
It's better to show all the stages with the number of join fields that they declared.
| } | ||
| } | ||
| } | ||
|
|
| Iterator<String> fieldIter = otherKey.getFields().iterator(); | ||
| for (String key1Field : key1.getFields()) { | ||
| // guaranteed to be non-null because of earlier validation | ||
| Schema key1Schema = schema1.getField(key1Field).getSchema(); |
There was a problem hiding this comment.
Good refactor these three lines as a private method:
Schema.Type key1Type = getFieldType(schema1, key1Field);
Schema.Type otherKeyType = getFieldType(otherSchema, fieldIter.next());
if (key1Type != otherKeyType) {
...
}
Add a new set of classes for AutoJoiner, which can be used instead of the current Joiner interface. This new API leaves all of the implementation details up to the application, which will allow the app to perform the join in better ways. For example, in the Spark program, it will allow using broadcast joins. Plugin developers are responsible for returning a JoinDefinition based on information about incoming stages. This change includes the JoinDefinition as well as all the classes required to create a definition. It also includes validation logic to make sure the plugin cannot create a definition that tries to join on a field that doesn't exist, or tries to join on fields that have mismatched types, or any other type of error.
albertshau
force-pushed
the
feature_release/CDAP-16708-new-joiner-api
branch
from
519f129
to
a568b1c
Compare
| String otherField = fieldIter.next(); | ||
| Schema otherFieldSchema = getNonNullableFieldSchema(otherSchema, otherField); | ||
|
|
||
| if (key1Schema.getType() != otherFieldSchema.getType()) { |
There was a problem hiding this comment.
should we also compare logical types?
There was a problem hiding this comment.
I thought about that and wavered back and forth.
In the end, I decided to allow it because it is actually able to join if the physical types are the same, regardless of the logical types. It probably makes sense for the plugin to check this and disallow, but didn't want to restrict it in the app.
| * | ||
| * <pre> | ||
| * {@code | ||
| * JoinDefinition define(AutoJoinerContext context) { |
Add a new set of classes for AutoJoiner, which can be used instead
of the current Joiner interface. This new API leaves all of the
implementation details up to the application, which will allow the
app to perform the join in better ways. For example, in the Spark
program, it will allow using broadcast joins.
Plugin developers are responsible for returning a JoinDefinition
based on information about incoming stages.
This change includes the JoinDefinition as well as all the classes
required to create a definition. It also includes validation logic
to make sure the plugin cannot create a definition that tries to
join on a field that doesn't exist, or tries to join on fields
that have mismatched types, or any other type of error.