fix(sql): fix a bug when planning semi- or antijoins

[aalexandrov]

• Closes exclude_using_columns might wrongly retain columns from a projected join input #20989.

Rationale for this change

The planner should be consistent with the expected SQL behavior—swapping the names of tables that have identical structure in a SQL query should not affect the schema for that query.

What changes are included in this PR?

• A fix in the exclude_using_columns helper method in datafusion/expr/src/utils.rs that ensures that we don't retain columns from the projected side when deciding which USING columns to exclude and which to retain on top of semi- or antijoins.
• Regression tests for the change in test_using_join_wildcard_schema_semi_anti.

Are these changes tested?

• Added a regression test.

Are there any user-facing changes?

Yes, the change is user facing, but I doubt that this behavior is expected and is documented anywhere.
If existing docs need to be updated, please point me to the concrete places and I can take a look.

[neilconway]

Thanks for taking this! Overall looks good to me.

[neilconway]

This test (and the other LEFT ANTI JOIN test) should succeed without this PR, right? Because the correct side of the join happens to be picked. Can we change this (or add a new test variant) that checks that the behavior has changed for LEFT joins?

[aalexandrov]

Done.

[neilconway]

It took me a while to wrap my head around the logic here; what do you think of this instead?

Suggested change

	/// When a JOIN has a USING clause, the join columns appear in the output
	/// schema once per side (for inner/outer joins) or once total (for semi/anti
	/// joins). An unqualified wildcard should include each USING column only once.
	/// This function returns the duplicate columns that should be excluded.
	fn exclude_using_columns(plan: &LogicalPlan) -> Result<HashSet<Column>> {
	let output_columns: HashSet<_> = plan.schema().columns().iter().cloned().collect();
	let mut excluded = HashSet::new();
	for cols in plan.using_columns()? {
	// `using_columns()` returns join columns from both sides regardless
	// of join type. For semi/anti joins, only one side's columns appear
	// in the output schema. Filter to output columns so that columns
	// from the non-output side don't participate in deduplication and
	// displace real output columns.
	let mut cols: Vec<_> = cols
	.into_iter()
	.filter(|c| output_columns.contains(c))
	.collect();
	// Sort so we keep the same qualified column, regardless of HashSet
	// iteration order.
	cols.sort();
	let mut seen = HashSet::new();
	for col in cols {
	if !seen.insert(col.name.clone()) {
	excluded.insert(col);
	}
	}
	}
	Ok(excluded)
	}

[aalexandrov]

I don't have a strong preference here—in the initial PR I was trying to do the minimal amount of changes to the existing method definition to fix the error.

I did some cosmetic changes to your suggestion and added it to the PR. I went with this for-loop body

if seen_names.contains(col.name.as_str()) {
    excluded.insert(col); // exclude columns with already seen name
} else {
    seen_names.insert(col.name.clone()); // mark column name as seen
}

because seen_names.contains(col.name.clone()) will create unnecessary String clones for ~50% of the cols entries.