[Data] [Do not merge] Check column naming removal impact on TPCH queries

[rayhhome]

Description

This PR reverts #63384 to measure TPCH release tests runtime.

[gemini-code-assist]

Code Review

This pull request implements projection pushdown with column renaming for Ray Data, updating datasources like Iceberg and Parquet to handle column renames directly at the read stage. It also adjusts predicate pushdown rules to correctly translate column namespaces. The review feedback highlights critical issues in the predicate translation logic within predicate_pushdown.py, specifically noting that the pushdown step should use the inverse rename map while the residual step should use the original rename map. Additionally, it is recommended to gracefully fall back by keeping the Project operator instead of raising a RuntimeError when the schema is unavailable in projection_pushdown.py.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The predicate substitution here is using rename_map directly, which maps original column names to renamed column names (e.g., {'original': 'renamed'}). However, the incoming predicate_expr is defined in terms of the renamed column names. To translate the predicate back to the original column namespace that the scanner understands, we must use the inverse of the rename map (i.e., {'renamed': 'original'}). Using rename_map directly will result in no substitution occurring, causing the scanner to receive renamed column names that do not exist in the physical schema.

Suggested change

	rename_map = input_op.get_column_renames()
	if rename_map:
	predicate_expr = cls._substitute_predicate_columns(
	predicate_expr, rename_map
	)
	rename_map = input_op.get_column_renames()
	if rename_map:
	inverse_rename_map = {new: old for old, new in rename_map.items()}
	predicate_expr = cls._substitute_predicate_columns(
	predicate_expr, inverse_rename_map
	)

[gemini-code-assist]

The residual predicate substitution here is using inverse_rename_map (which maps renamed column names to original column names). However, the residual filter needs to be rebound back to the renamed column namespace because the downstream operator will apply the column renames at runtime. Therefore, we should use the original rename_map (which maps original column names to renamed column names) instead of the inverse map.

            if rename_map:
                residual = cls._substitute_predicate_columns(
                    residual, rename_map
                )

[gemini-code-assist]

Raising a RuntimeError when the schema is unavailable can crash the entire query optimization phase. For custom or legacy datasources that do not expose schema information, it is safer to gracefully fall back by keeping the Project operator on top of the read operator (i.e., returning current_project without pushing down the rename).

Suggested change

	else:
	# Cannot determine available columns - this shouldn't happen in practice
	# for properly implemented datasources. Rather than guessing, raise an error.
	raise RuntimeError(
	f"Cannot apply rename operation: schema unavailable for input operator "
	f"{input_op}. This may indicate a legacy datasource that doesn't properly "
	f"expose schema information."
	)
	else:
	return current_project

[github-actions]

This pull request has been automatically marked as stale because it has not had
any activity for 14 days. It will be closed in another 14 days if no further activity occurs.
Thank you for your contributions.

You can always ask for help on our discussion forum or Ray's public slack channel.

If you'd like to keep this open, just leave any comment, and the stale label will be removed.