Add input_file_name built-in function

[nkarpov]

Is your feature request related to a problem or challenge?

It's useful to project the source input file of a data row to support file aware operations, for example for storage frameworks (delta-io/delta-rs#850). This is a built-in function in Spark, for example, https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.input_file_name.html

There was prior work before the repository split but it looks to have lost momentum:

apache/arrow#9944
apache/arrow#9976
apache/arrow#18601

Based on the conversations in the prior PRs and issues it looks like there was consensus that this feature should live in datafusion as opposed to arrow, so creating an issue here.

Describe the solution you'd like

A built-in function supported for both SQL and DataFrame APIs input_file_name() that returns a string of the file from which the row was originally scanned.

Describe alternatives you've considered

No response

Additional context

No response

[SteveLauC]

I am interested in implementing this as requested by discussion#7979, just checked the the previous PRs, and here are my thoughts:

1. What is the correct semantics of this input_file_name() function

   1. Return all the files registered by a table ~~
   2. The file that a specific row comes from (at runtime/execution time)

   arrow#9944 seems to choose option 1 if I understand correctly, in my use case, I would like to have option 2.

   Update: after taking a detailed look at that PR, I realized it implemented the option 2 by storing the filename in the Schema.metadata

2. For option 1, this info is stored in different file's ExecutionPlan node, we can fetch it after generating the physical plan

3. For option 2, this info will be available when different file's XXXOpener type actually opens a file (whether on the local file system or object storage)

---

Friendly ping @alamb and @jorgecarleitao since you guys were the reviewer of the previous PR, I would like to hear your thoughts! :)

And, haven't made any serious contributions to DataFusion, guidance would be hightly appreciated!

[alamb]

What is the correct semantics of this input_file_name() function

I do not know. However, the use case of " that returns a string of the file from which the row was originally scanned" suggests it would be i in your list: "Return all the files registered by a table"

The spark docs say

Creates a string column for the file name of the current Spark task.

I am not sure how that maps to what file is processed (aka does the same task process multiple input files?)

As for implementation, perhaps it would be possible to model how partition columns are injected, though as you will find that is non trivial to implement.

[alamb]

My "gut feeling" is that this will be a very complicated feature to implement with the existing Listing table provider.

[adriangb]

copying #18494 (comment) over here:

maybe we can implement option ii (The file that a specific row comes from) by use an extension type marker and fill it in via FIleScanConfig or ParquetOpener, similar to what arrow-rs does at the row group level in apache/arrow-rs#9117

[alamb]

I also think the ListingTableProvider is in signficantly better shape now. If we could model the file name as a "partition column" it may nicely just work

[ethan-tyler]

+1 - This is the “file identity” half we need for file aware table format semantics.

We hit an issue in delta-rs where DF coalesced batches across file boundaries (fixed in next-scan), which surfaced that relying on implicit stream ordering is fragile for DV semantics and log replay. delta-io/delta-rs#4115.

input_file_name() can get us to a stable (file, …) key. For fully order insensitive DV reads/writes we would also need (…, row_position) (0-based file ordinal)See #13261.

This would be a big help for us on the delta-rs side. Happy to take this on or help however needed. Lmk what would be most useful.

[alamb]

@ethan-tyler it would be great if you wanted to work on this

I would suggest: Sketching out a PR to get the feature to work (basically a plumbing exercise) and then we can figure out how to break it down into smaller PRs for review

I would personally recommend starting with a slt test that shows what you want to do -- like

CREATE EXTERNAL TABLE t AS ...

-- not sure what the syntax would be here
SELECT col1, __filename from t

[adriangb]

May I suggest that since scans can now receive expressions we make a function called filename() that is implemented to return null by default but that scans swap out with the literal filename if they see it in the projection?

Could be implemented as a “trivial expression” that we’ve been discussing Andrew. But there might also be too many implementation challenges getting it down into the scan.

[alamb]

A function that gets pushed down (and rewritten by the file opener (which knows the file name)) sounds like a great idea to me

[adriangb]

A function that gets pushed down (and rewritten by the file opener (which knows the file name)) sounds like a great idea to me

As you know… the issue will be making sure it’s always pushed down. select min(col), max(col) from t group by filename() order by filename() should ideally work.

[ethan-tyler]

@alamb - Awesome! Happy to take this on, appreciate the endorsement!

Starting with an SLT for Parquet external tables:

CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '...';
SELECT col1, input_file_name() FROM t;

I’d prefer keeping the semantics “Spark-like” with input_file_name() , can adjust if we prefer filename().

@adriangb - Love it :) Trivial expression that defaults to NULL, rewritten by file opener. Cost paid only when referenced and no schema mutation. Composes well with (file, row_position) direction for DV semantics (#13261).

Good point on ensuring pushdown for GROUP BY filename() / ORDER BY filename(). May need the rewrite early in planning so downstream operators see the resolved column. I'll reach out if that gets tricky.

I'll prototype the rewrite in Parquet scan first (per-row literal, likely dictionary encoded). Once SLT passes, I'll open a draft PR to iterate on naming/semantics and gather feedback before splitting into reviewable pieces.

[adriangb]

Good point on ensuring pushdown for GROUP BY filename() / ORDER BY filename(). May need the rewrite early in planning so downstream operators see the resolved column. I'll reach out if that gets tricky.

Oh it will! Andrew and I are working on exactly this problem but for struct field access. I’d say let’s not duplicate effort in this regard. So maybe like you say start with “what if this expression is in a projection right above the ParquetSource” and then we can work on making sure it gets there from the GroupBy. Some background: #19538. I need to do work on this PR but maybe I can also tag you for review on related work?

[ethan-tyler]

@adriangb - Perfect, happy to start with the projection above ParquetSource case and let the broader pushdown work land separately. Will check out #19538 for context.

You can always tag me on reviews, happy to help!