[VL] Fix input_file_name results in empty string

[zml1206]

What changes were proposed in this pull request?

The Spark implementation of input_file_name uses a thread local to stash the file name and retrieve it from the function.

If there is a transformer node between project input_file_name and scan, the result of input_file_name is an empty string.
For example, read delta lake table need union checkpoint parquet file and json file, then order by input_file_name to get parquet data files, it will get wrong parquet file list.
So we should push down input_file_name to transformer scan or add fallback project before fallback scan

How was this patch tested?

UT

[github-actions]

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

• Other pull requests

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[zml1206]

Run Gluten Clickhouse CI

[zhztheplayer]

cc @gaoyangxiaozhu thanks

[zml1206]

cc @gaoyangxiaozhu Can you help take a look if you have time? Thank you.

[zml1206]

@JkSelf Can you help take a look if you have time? Thank you.

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[zhztheplayer]

Is there real case that we see a BatchScanExecTransformer in OffloadProject? Since OffloadOthers is executed after OffloadProject.

[zml1206]

I agree, this is the previous PR code, should it be removed in this PR?

[liuneng1994]

I encountered the same problem. In delta, input_file_name and monotonically_increasing_id are used at the same time. Monotonically_increasing_id is a state function, which is not easy to support natively. The existing logic will lose the fallback tag of the child of input_file_name, resulting in incorrect fallback.

example plan

BroadcastHashJoin [l_orderkey#363L], [l_orderkey#906L], Inner, BuildRight, false
:- Project [l_orderkey#363L]
:  +- Filter isnotnull(l_orderkey#363L)
:     +- Filter UDF()
:        +- Scan ExistingRDD mergeMaterializedSource[l_orderkey#363L,l_partkey#364L,l_suppkey#365L,l_linenumber#366L,l_quantity#367,l_extendedprice#368,l_discount#369,l_tax#370,l_returnflag#903,l_linestatus#372,l_shipdate#373,l_commitdate#374,l_receiptdate#375,l_shipinstruct#376,l_shipmode#377,l_comment#378]
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=1263]
   +- Filter isnotnull(l_orderkey#906L)
      +- Project [l_orderkey#906L, _row_id_#1183L, input_file_name() AS _file_name_#1201]
         +- Project [l_orderkey#906L, monotonically_increasing_id() AS _row_id_#1183L]
            +- FileScan mergetree [l_orderkey#906L] Batched: true, DataFilters: [], Format: MergeTree, Location: TahoeBatchFileIndex(1 paths)[file:/home/admin1/github/gazelle-jni/backends-clickhouse/target/scal..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<l_orderkey:bigint>

[zml1206]

@zhztheplayer Thank you for review. This pr can be reviewed later. I am reconstructing this logic.

1. Add a new before offload rule to push down the input file to before scan, and add an additional project. The output is the output+inputfile of the scan.
2. Normal offload project, the inputfile project before scan will not be offloaded.
3. Add a new rule after offload. If scan offloads, push the inputfile to scan and delete the project.
   In this way, the overall logic is clearer, the offload logic will not be invaded, and RAS can also be used directly.
   cc @liuneng1994

[zhztheplayer]

@zml1206 OK, so let's mark the PR as draft before it's ready?

[zml1206]

new PR #7124