Add try_new for LogicalPlan::Join

[kumarlokesh]

Which issue does this PR close?

• Closes Add try_new for LogicalPlan::Join Join and others #14363.

Rationale for this change

What changes are included in this PR?

This PR adds a new Join::try_new() constructor method that:

• Automatically computes the correct schema based on join type

• Leverages the existing build_join_schema() function

• Handles all join types correctly (Inner, Left, Right, Full, Semi, Anti, and LeftMark)

• Additionally, it updates the LogicalPlanBuilder:

  • cross_join()
  • join_with_expr_keys()
  • join_using()

to use the new try_new method instead of manual schema computation.

Are these changes tested?

Are there any user-facing changes?

[ctsk]

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.

datafusion/datafusion/expr/src/logical_plan/builder.rs

Lines 1082 to 1099 in d1cd58c

	let join_schema =
	build_join_schema(self.plan.schema(), right.schema(), &join_type)?;
	// Inner type without join condition is cross join
	if join_type != JoinType::Inner && on.is_empty() && filter.is_none() {
	return plan_err!("join condition should not be empty");
	}
	Ok(Self::new(LogicalPlan::Join(Join {
	left: self.plan,
	right: Arc::new(right),
	on,
	filter,
	join_type,
	join_constraint: JoinConstraint::On,
	schema: DFSchemaRef::new(join_schema),
	null_equals_null,
	})))

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!

[comphead]

lgtm thanks @kumarlokesh