The implementation looks good to me. Two points:

1. The testing seems quite verbose - I briefly browsed around and think DFSchema::matches_arrow_schema can help you shorten it.
2. There are might be some more places where this is applicable:
   

   datafusion/datafusion/expr/src/logical_plan/plan.rs

   Lines 663 to 683 in d1cd58c

   	let schema =
   	build_join_schema(left.schema(), right.schema(), &join_type)?;
   	let new_on: Vec<_> = on
   	.into_iter()
   	.map(|equi_expr| {
   	// SimplifyExpression rule may add alias to the equi_expr.
   	(equi_expr.0.unalias(), equi_expr.1.unalias())
   	})
   	.collect();
   	Ok(LogicalPlan::Join(Join {
   	left,
   	right,
   	join_type,
   	join_constraint,
   	on: new_on,
   	filter,
   	schema: DFSchemaRef::new(schema),
   	null_equals_null,
   	}))

   
   

   datafusion/datafusion/expr/src/logical_plan/plan.rs

   Lines 3771 to 3783 in d1cd58c

   	let join_schema =
   	build_join_schema(left.schema(), right.schema(), &original_join.join_type)?;
   	Ok(Join {
   	left,
   	right,
   	on,
   	filter: original_join.filter.clone(),
   	join_type: original_join.join_type,
   	join_constraint: original_join.join_constraint,
   	schema: Arc::new(join_schema),
   	null_equals_null: original_join.null_equals_null,
   	})

   
   

   datafusion/datafusion/optimizer/src/eliminate_cross_join.rs

   Lines 317 to 332 in d1cd58c

   	let join_schema = Arc::new(build_join_schema(
   	left_input.schema(),
   	right_input.schema(),
   	&JoinType::Inner,
   	)?);
   	return Ok(LogicalPlan::Join(Join {
   	left: Arc::new(left_input),
   	right: Arc::new(right_input),
   	join_type: JoinType::Inner,
   	join_constraint: JoinConstraint::On,
   	on: join_keys,
   	filter: None,
   	schema: join_schema,
   	null_equals_null: false,
   	}));

   
   

   datafusion/datafusion/optimizer/src/eliminate_cross_join.rs

   Lines 339 to 354 in d1cd58c

   	let join_schema = Arc::new(build_join_schema(
   	left_input.schema(),
   	right.schema(),
   	&JoinType::Inner,
   	)?);
   	Ok(LogicalPlan::Join(Join {
   	left: Arc::new(left_input),
   	right: Arc::new(right),
   	schema: join_schema,
   	on: vec![],
   	filter: None,
   	join_type: JoinType::Inner,
   	join_constraint: JoinConstraint::On,
   	null_equals_null: false,
   	}))

Could be also applied here, but would change the behaviour in the error case - I don't know if that matters.

Note that all of these are mere suggestions, I am not very familiar with this part of datafusion and apologize if any of them are wrong.

Cheers!