@Test public void testJoin() { Schema firstSchema = new Schema.Builder().addColumnString("keyColumn").addColumnsInteger("first0", "first1").build(); Schema secondSchema = new Schema.Builder().addColumnString("keyColumn").addColumnsInteger("second0").build(); List<List<Writable>> first = new ArrayList<>(); first.add(Arrays.asList((Writable) new Text("key0"), new IntWritable(0), new IntWritable(1))); first.add(Arrays.asList((Writable) new Text("key1"), new IntWritable(10), new IntWritable(11))); List<List<Writable>> second = new ArrayList<>(); second.add(Arrays.asList((Writable) new Text("key0"), new IntWritable(100))); second.add(Arrays.asList((Writable) new Text("key1"), new IntWritable(110))); Join join = new Join.Builder(Join.JoinType.Inner).setJoinColumns("keyColumn") .setSchemas(firstSchema, secondSchema).build(); List<List<Writable>> expected = new ArrayList<>(); expected.add(Arrays.asList((Writable) new Text("key0"), new IntWritable(0), new IntWritable(1), new IntWritable(100))); expected.add(Arrays.asList((Writable) new Text("key1"), new IntWritable(10), new IntWritable(11), new IntWritable(110))); Schema joinedSchema = join.getOutputSchema(); assertEquals(4, joinedSchema.numColumns()); assertEquals(Arrays.asList("keyColumn", "first0", "first1", "second0"), joinedSchema.getColumnNames()); assertEquals(Arrays.asList(ColumnType.String, ColumnType.Integer, ColumnType.Integer, ColumnType.Integer), joinedSchema.getColumnTypes()); expected = new ArrayList<>(); expected.add(Arrays.asList((Writable) new Text("key0"), new IntWritable(0), new IntWritable(1), NullWritable.INSTANCE)); expected.add(Arrays.asList((Writable) new Text("key1"), new IntWritable(10), new IntWritable(11), NullWritable.INSTANCE)); for (int i = 0; i < first.size(); i++) { List<Writable> out = join.joinExamples(first.get(i), null); assertEquals(expected.get(i), out); } expected = new ArrayList<>(); expected.add(Arrays.asList((Writable) new Text("key0"), NullWritable.INSTANCE, NullWritable.INSTANCE, new IntWritable(100))); expected.add(Arrays.asList((Writable) new Text("key1"), NullWritable.INSTANCE, NullWritable.INSTANCE, new IntWritable(110))); for (int i = 0; i < first.size(); i++) { List<Writable> out = join.joinExamples(null, second.get(i)); assertEquals(expected.get(i), out); } }
private Join(Builder builder) { this.joinType = builder.joinType; this.leftSchema = builder.leftSchema; this.rightSchema = builder.rightSchema; this.joinColumnsLeft = builder.joinColumnsLeft; this.joinColumnsRight = builder.joinColumnsRight; if (joinType == null) throw new IllegalArgumentException("Join type cannot be null"); if (leftSchema == null) throw new IllegalArgumentException("Left schema cannot be null"); if (rightSchema == null) throw new IllegalArgumentException("Right schema cannot be null"); if (joinColumnsLeft == null || joinColumnsLeft.length == 0) { throw new IllegalArgumentException("Invalid left join columns: " + (joinColumnsLeft == null ? null : Arrays.toString(joinColumnsLeft))); } if (joinColumnsRight == null || joinColumnsRight.length == 0) { throw new IllegalArgumentException("Invalid right join columns: " + (joinColumnsRight == null ? null : Arrays.toString(joinColumnsRight))); } for (String leftCol : joinColumnsLeft) { if (!leftSchema.hasColumn(leftCol)) { throw new IllegalArgumentException("Cannot perform join: left join column \"" + leftCol + "\" does not exist in left schema. All columns in left schema: " + leftSchema.getColumnNames()); } } for (String rightCol : joinColumnsRight) { if (!rightSchema.hasColumn(rightCol)) { throw new IllegalArgumentException("Cannot perform join: right join column \"" + rightCol + "\" does not exist in right schema. All columns in right schema: " + rightSchema.getColumnNames()); } } }
Join implements Serializable { private Join(Builder builder) { this.joinType = builder.joinType; this.leftSchema = builder.leftSchema; this.rightSchema = builder.rightSchema; this.joinColumnsLeft = builder.joinColumnsLeft; this.joinColumnsRight = builder.joinColumnsRight; if (joinType == null) throw new IllegalArgumentException("Join type cannot be null"); if (leftSchema == null) throw new IllegalArgumentException("Left schema cannot be null"); if (rightSchema == null) throw new IllegalArgumentException("Right schema cannot be null"); if (joinColumnsLeft == null || joinColumnsLeft.length == 0) { throw new IllegalArgumentException("Invalid left join columns: " + (joinColumnsLeft == null ? null : Arrays.toString(joinColumnsLeft))); } if (joinColumnsRight == null || joinColumnsRight.length == 0) { throw new IllegalArgumentException("Invalid right join columns: " + (joinColumnsRight == null ? null : Arrays.toString(joinColumnsRight))); } for (String leftCol : joinColumnsLeft) { if (!leftSchema.hasColumn(leftCol)) { throw new IllegalArgumentException("Cannot perform join: left join column \"" + leftCol + "\" does not exist in left schema. All columns in left schema: " + leftSchema.getColumnNames()); } } for (String rightCol : joinColumnsRight) { if (!rightSchema.hasColumn(rightCol)) { throw new IllegalArgumentException("Cannot perform join: right join column \"" + rightCol + "\" does not exist in right schema. All columns in right schema: " + rightSchema.getColumnNames()); } } } }
Join implements Serializable { private Join(Builder builder) { this.joinType = builder.joinType; this.leftSchema = builder.leftSchema; this.rightSchema = builder.rightSchema; this.joinColumnsLeft = builder.joinColumnsLeft; this.joinColumnsRight = builder.joinColumnsRight; if (joinType == null) throw new IllegalArgumentException("Join type cannot be null"); if (leftSchema == null) throw new IllegalArgumentException("Left schema cannot be null"); if (rightSchema == null) throw new IllegalArgumentException("Right schema cannot be null"); if (joinColumnsLeft == null || joinColumnsLeft.length == 0) { throw new IllegalArgumentException("Invalid left join columns: " + (joinColumnsLeft == null ? null : Arrays.toString(joinColumnsLeft))); } if (joinColumnsRight == null || joinColumnsRight.length == 0) { throw new IllegalArgumentException("Invalid right join columns: " + (joinColumnsRight == null ? null : Arrays.toString(joinColumnsRight))); } for (String leftCol : joinColumnsLeft) { if (!leftSchema.hasColumn(leftCol)) { throw new IllegalArgumentException("Cannot perform join: left join column \"" + leftCol + "\" does not exist in left schema. All columns in left schema: " + leftSchema.getColumnNames()); } } for (String rightCol : joinColumnsRight) { if (!rightSchema.hasColumn(rightCol)) { throw new IllegalArgumentException("Cannot perform join: right join column \"" + rightCol + "\" does not exist in right schema. All columns in right schema: " + rightSchema.getColumnNames()); } } } private  Join(Builder builder); }
Join implements Serializable { private Join(Builder builder) { this.joinType = builder.joinType; this.leftSchema = builder.leftSchema; this.rightSchema = builder.rightSchema; this.joinColumnsLeft = builder.joinColumnsLeft; this.joinColumnsRight = builder.joinColumnsRight; if (joinType == null) throw new IllegalArgumentException("Join type cannot be null"); if (leftSchema == null) throw new IllegalArgumentException("Left schema cannot be null"); if (rightSchema == null) throw new IllegalArgumentException("Right schema cannot be null"); if (joinColumnsLeft == null || joinColumnsLeft.length == 0) { throw new IllegalArgumentException("Invalid left join columns: " + (joinColumnsLeft == null ? null : Arrays.toString(joinColumnsLeft))); } if (joinColumnsRight == null || joinColumnsRight.length == 0) { throw new IllegalArgumentException("Invalid right join columns: " + (joinColumnsRight == null ? null : Arrays.toString(joinColumnsRight))); } for (String leftCol : joinColumnsLeft) { if (!leftSchema.hasColumn(leftCol)) { throw new IllegalArgumentException("Cannot perform join: left join column \"" + leftCol + "\" does not exist in left schema. All columns in left schema: " + leftSchema.getColumnNames()); } } for (String rightCol : joinColumnsRight) { if (!rightSchema.hasColumn(rightCol)) { throw new IllegalArgumentException("Cannot perform join: right join column \"" + rightCol + "\" does not exist in right schema. All columns in right schema: " + rightSchema.getColumnNames()); } } } private  Join(Builder builder); Schema getOutputSchema(); List<Writable> joinExamples(List<Writable> leftExample, List<Writable> rightExample); }
Join implements Serializable { private Join(Builder builder) { this.joinType = builder.joinType; this.leftSchema = builder.leftSchema; this.rightSchema = builder.rightSchema; this.joinColumnsLeft = builder.joinColumnsLeft; this.joinColumnsRight = builder.joinColumnsRight; if (joinType == null) throw new IllegalArgumentException("Join type cannot be null"); if (leftSchema == null) throw new IllegalArgumentException("Left schema cannot be null"); if (rightSchema == null) throw new IllegalArgumentException("Right schema cannot be null"); if (joinColumnsLeft == null || joinColumnsLeft.length == 0) { throw new IllegalArgumentException("Invalid left join columns: " + (joinColumnsLeft == null ? null : Arrays.toString(joinColumnsLeft))); } if (joinColumnsRight == null || joinColumnsRight.length == 0) { throw new IllegalArgumentException("Invalid right join columns: " + (joinColumnsRight == null ? null : Arrays.toString(joinColumnsRight))); } for (String leftCol : joinColumnsLeft) { if (!leftSchema.hasColumn(leftCol)) { throw new IllegalArgumentException("Cannot perform join: left join column \"" + leftCol + "\" does not exist in left schema. All columns in left schema: " + leftSchema.getColumnNames()); } } for (String rightCol : joinColumnsRight) { if (!rightSchema.hasColumn(rightCol)) { throw new IllegalArgumentException("Cannot perform join: right join column \"" + rightCol + "\" does not exist in right schema. All columns in right schema: " + rightSchema.getColumnNames()); } } } private  Join(Builder builder); Schema getOutputSchema(); List<Writable> joinExamples(List<Writable> leftExample, List<Writable> rightExample);  }