[Data] Fix renamed columns to be appropriately dropped from the ouput

[alexeykudinkin]

Description

This change addresses the issues that currently upon column renaming we're not removing original columns.

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request refactors and fixes projection pushdown logic, specifically how renamed columns are handled during the fusion of Project operators. The changes introduce a distinction between alias and rename operations, with rename now correctly signaling that the source column should be dropped. The logic for fusing consecutive projections has been significantly simplified and corrected, especially for composition cases involving star() expressions. The _ColumnRewriter has been replaced with a much simpler _ColumnRefRebindingVisitor, improving maintainability. Overall, these are excellent changes that make the projection fusion logic more robust and easier to understand. I've found one issue in a new test case where an assertion could be unreliable.

[gemini-code-assist]

The assertion select_op.exprs == [...] is unreliable for comparing lists of expression objects. The __eq__ method on Expr is overloaded to build new expressions (e.g., col('a') == 1 creates a filter expression), not to check for equality. This means the assertion doesn't check for structural equality and might pass incorrectly or fail unexpectedly.

To ensure the test is robust, you should explicitly use structurally_equals for each expression in the list.

Suggested change

	assert select_op.exprs == [
	# TODO fix (renaming doesn't remove prev columns)
	col("sepal.length").alias("length"),
	col("petal.width").alias("width"),
	]
	expected_exprs = [
	col("sepal.length").alias("length"),
	col("petal.width").alias("width"),
	]
	assert len(select_op.exprs) == len(expected_exprs)
	assert select_op.exprs[0].structurally_equals(expected_exprs[0])
	assert select_op.exprs[1].structurally_equals(expected_exprs[1])

[cursor]

Bug: Projection Fusion Fails on Renamed Columns

The _get_col_refs_removed_by_renaming function incorrectly marks source columns as "removed" for any rename operation, even if the column is still present in the projection's output. This can lead to "column not found" errors during projection fusion.

 

[cursor]

Bug: Self-Rename Bug in Column Projection

The _extract_simple_rename function incorrectly marks columns as removed during self-renames (e.g., col("a").rename("a")). This occurs because the source_name != target_name check was removed, causing projection fusion to incorrectly report missing columns.

 

[goutamvenkat-anyscale]

nit: _ColumnRefSubstitutionVisitor is a little clearer

[alexeykudinkin]

Agreed. Will defer to follow-up tho

[goutamvenkat-anyscale]

Wait this should be a public api.