Skip to content

Commit f0e89d4

Browse files
committed
Code review
Signed-off-by: Adam Gutglick <adamgsal@gmail.com>
1 parent 5cc7e1a commit f0e89d4

7 files changed

Lines changed: 144 additions & 34 deletions

File tree

datafusion/datasource-parquet/src/source.rs

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -679,6 +679,8 @@ impl FileSource for ParquetSource {
679679
) -> datafusion_common::Result<Option<Arc<dyn FileSource>>> {
680680
let mut source = self.clone();
681681

682+
// If there's no reference to `FileRowIndexFunc` in the projection, we can just merge
683+
// both projections as-is, there's no need to modify the projection first.
682684
if !projection.iter().any(|projection_expr| {
683685
expr_references_scalar_udf::<FileRowIndexFunc>(&projection_expr.expr)
684686
}) {
@@ -1014,9 +1016,14 @@ impl FileSource for ParquetSource {
10141016
}
10151017
}
10161018

1019+
/// Returns the a [`TableSchema`] containing a [`RowNumber`] virtual column and a [`Column`] expression referencing its row index column.
1020+
/// The expression is then merged into a projection.
1021+
///
1022+
/// - If the schema already has a virtual column with the [`RowNumber`] type, it returns the schema unchanged.
1023+
/// - If the schema doesn't have the appropriate virtual column, it returns a modified schema with the virtual column appended to it.
10171024
fn table_schema_with_row_index_col(table_schema: &TableSchema) -> (TableSchema, Column) {
1018-
let virtual_offset = table_schema.file_schema().fields().len()
1019-
+ table_schema.table_partition_cols().len();
1025+
// If we can find a virtual column with the `RowNumber` type, we just return the schema
1026+
// and create the appropriate `column` we're going to use
10201027
if let Some((idx, field)) =
10211028
table_schema
10221029
.virtual_columns()
@@ -1028,6 +1035,9 @@ fn table_schema_with_row_index_col(table_schema: &TableSchema) -> (TableSchema,
10281035
.is_some_and(|name| name == RowNumber::NAME)
10291036
})
10301037
{
1038+
let virtual_offset = table_schema.file_schema().fields().len()
1039+
+ table_schema.table_partition_cols().len();
1040+
10311041
return (
10321042
table_schema.clone(),
10331043
Column::new(field.name(), virtual_offset + idx),

datafusion/functions/src/core/file_row_index.rs

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
2020
use arrow::datatypes::DataType;
2121
use datafusion_common::utils::take_function_args;
22-
use datafusion_common::{Result, ScalarValue};
22+
use datafusion_common::{Result, exec_err};
2323
use datafusion_doc::Documentation;
2424
use datafusion_expr::{
2525
ColumnarValue, ExpressionPlacement, ScalarFunctionArgs, ScalarUDFImpl, Signature,
@@ -30,8 +30,8 @@ use datafusion_macros::user_doc;
3030
/// Scalar UDF implementation for `file_row_index()`.
3131
///
3232
/// File sources that can expose per-file row indexes rewrite this placeholder
33-
/// function into a source-provided physical expression. Its fallback
34-
/// evaluation returns NULL because there is no file context outside a scan.
33+
/// function into a source-provided physical expression. Direct evaluation
34+
/// returns an error because there is no file context outside a scan.
3535
#[user_doc(
3636
doc_section(label = "Other Functions"),
3737
description = r#"Returns the zero-based row offset within the source file
@@ -40,8 +40,8 @@ that produced the current row.
4040
The value is scoped to one file, so rows from different files in the same scan
4141
can have the same row index. This function is intended to be rewritten at
4242
file-scan time. If the input file is not known (for example, if this function
43-
is evaluated outside a file scan, or was not pushed down into one), this
44-
function returns NULL.
43+
is evaluated outside a file scan, or was not pushed down into one), direct
44+
evaluation returns an error.
4545
"#,
4646
syntax_example = "file_row_index()",
4747
sql_example = r#"```sql
@@ -83,7 +83,7 @@ impl ScalarUDFImpl for FileRowIndexFunc {
8383

8484
fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
8585
let [] = take_function_args(self.name(), args.args)?;
86-
Ok(ColumnarValue::Scalar(ScalarValue::Int64(None)))
86+
exec_err!("file_row_index() is source dependent and cannot be evaluated directly")
8787
}
8888

8989
fn placement(&self, _args: &[ExpressionPlacement]) -> ExpressionPlacement {

datafusion/physical-expr-adapter/src/lib.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,5 +30,5 @@ pub use schema_rewriter::{
3030
BatchAdapter, BatchAdapterFactory, DefaultPhysicalExprAdapter,
3131
DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
3232
expr_references_scalar_udf, replace_columns_with_literals,
33-
rewrite_file_row_index_expr, rewrite_file_row_index_projection, rewrite_scalar_udf,
33+
rewrite_file_row_index_expr, rewrite_file_row_index_projection,
3434
};

datafusion/physical-expr-adapter/src/schema_rewriter.rs

Lines changed: 11 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@ pub fn expr_references_scalar_udf<T: ScalarUDFImpl>(
110110
/// The rewrite matches the concrete [`ScalarUDFImpl`] type rather than the
111111
/// function name. `replacement` is called with each matching
112112
/// [`ScalarFunctionExpr`] after its children have been rewritten.
113-
pub fn rewrite_scalar_udf<T, F>(
113+
fn rewrite_scalar_udf<T, F>(
114114
expr: Arc<dyn PhysicalExpr>,
115115
mut replacement: F,
116116
) -> Result<Arc<dyn PhysicalExpr>>
@@ -166,36 +166,28 @@ pub fn rewrite_file_row_index_projection(
166166
row_index_col: &Column,
167167
) -> Result<ProjectionExprs> {
168168
let mut base_exprs = base_projection.as_ref().to_vec();
169-
let row_index_projection_idx = row_index_projection_idx(&base_exprs, row_index_col);
170-
if row_index_projection_idx == base_exprs.len() {
169+
let row_index_projection_idx =
170+
base_projection.projected_column_position(row_index_col);
171+
172+
// If the column doesn't exist in the projection yet
173+
if row_index_projection_idx.is_none() {
171174
base_exprs.push(ProjectionExpr {
172175
expr: Arc::new(row_index_col.clone()),
173176
alias: row_index_col.name().to_owned(),
174177
});
175178
}
176179

177180
let rewritten_projection = projection.clone().try_map_exprs(|expr| {
178-
rewrite_file_row_index_expr(expr, row_index_col.name(), row_index_projection_idx)
181+
rewrite_file_row_index_expr(
182+
expr,
183+
row_index_col.name(),
184+
row_index_projection_idx.unwrap_or(base_exprs.len() - 1),
185+
)
179186
})?;
180187

181188
ProjectionExprs::new(base_exprs).try_merge(&rewritten_projection)
182189
}
183190

184-
fn row_index_projection_idx(
185-
projection: &[ProjectionExpr],
186-
row_index_col: &Column,
187-
) -> usize {
188-
projection
189-
.iter()
190-
.position(|projection| {
191-
projection
192-
.expr
193-
.downcast_ref::<Column>()
194-
.is_some_and(|column| column == row_index_col)
195-
})
196-
.unwrap_or(projection.len())
197-
}
198-
199191
/// Trait for adapting [`PhysicalExpr`] expressions to match a target schema.
200192
///
201193
/// This is used in file scans to rewrite expressions so that they can be

datafusion/physical-expr/src/projection.rs

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -725,6 +725,60 @@ impl ProjectionExprs {
725725
stats.column_statistics = column_statistics;
726726
Ok(stats)
727727
}
728+
729+
/// Returns the output position of `column` if this projection contains it.
730+
///
731+
/// This only matches projection expressions that are exactly [`Column`] expressions.
732+
/// Computed expressions, even if they reference `column`, do not match. The
733+
/// comparison uses [`Column`] equality, so both the name and index must match.
734+
/// If the same column appears more than once, this returns the first matching
735+
/// position.
736+
///
737+
/// # Example
738+
///
739+
/// ```rust
740+
/// use datafusion_common::ScalarValue;
741+
/// use datafusion_physical_expr::expressions::{Column, Literal};
742+
/// use datafusion_physical_expr::projection::{ProjectionExpr, ProjectionExprs};
743+
/// use std::sync::Arc;
744+
///
745+
/// let projection = ProjectionExprs::new([
746+
/// ProjectionExpr::new(Arc::new(Column::new("b", 1)), "b"),
747+
/// ProjectionExpr::new(
748+
/// Arc::new(Literal::new(ScalarValue::Int32(Some(42)))),
749+
/// "answer",
750+
/// ),
751+
/// ProjectionExpr::new(Arc::new(Column::new("a", 0)), "a"),
752+
/// ]);
753+
///
754+
/// assert_eq!(
755+
/// projection.projected_column_position(&Column::new("b", 1)),
756+
/// Some(0)
757+
/// );
758+
/// assert_eq!(
759+
/// projection.projected_column_position(&Column::new("a", 0)),
760+
/// Some(2)
761+
/// );
762+
///
763+
/// // The literal projection is not a Column expression.
764+
/// assert_eq!(
765+
/// projection.projected_column_position(&Column::new("answer", 1)),
766+
/// None
767+
/// );
768+
///
769+
/// // Columns not present in the projection also return None.
770+
/// assert_eq!(
771+
/// projection.projected_column_position(&Column::new("c", 2)),
772+
/// None
773+
/// );
774+
/// ```
775+
pub fn projected_column_position(&self, column: &Column) -> Option<usize> {
776+
self.iter().position(|expr| {
777+
expr.expr
778+
.downcast_ref::<Column>()
779+
.is_some_and(|projected| projected == column)
780+
})
781+
}
728782
}
729783

730784
/// Propagate column statistics through CAST projections. Other expressions
@@ -2212,6 +2266,43 @@ pub(crate) mod tests {
22122266
Schema::new(vec![field_0, field_1, field_2])
22132267
}
22142268

2269+
#[test]
2270+
fn test_projected_column_position_returns_output_position() {
2271+
let projection = ProjectionExprs::new([
2272+
ProjectionExpr::new(Arc::new(Column::new("col2", 2)), "col2"),
2273+
ProjectionExpr::new(Arc::new(Column::new("col0", 0)), "col0"),
2274+
]);
2275+
2276+
assert_eq!(
2277+
projection.projected_column_position(&Column::new("col2", 2)),
2278+
Some(0)
2279+
);
2280+
assert_eq!(
2281+
projection.projected_column_position(&Column::new("col0", 0)),
2282+
Some(1)
2283+
);
2284+
}
2285+
2286+
#[test]
2287+
fn test_projected_column_position_returns_none_for_non_column_or_missing() {
2288+
let projection = ProjectionExprs::new([
2289+
ProjectionExpr::new(
2290+
Arc::new(Literal::new(ScalarValue::Int64(Some(42)))),
2291+
"col1",
2292+
),
2293+
ProjectionExpr::new(Arc::new(Column::new("col0", 0)), "col0"),
2294+
]);
2295+
2296+
assert_eq!(
2297+
projection.projected_column_position(&Column::new("col1", 1)),
2298+
None
2299+
);
2300+
assert_eq!(
2301+
projection.projected_column_position(&Column::new("col2", 2)),
2302+
None
2303+
);
2304+
}
2305+
22152306
#[test]
22162307
fn test_stats_projection_columns_only() {
22172308
let source = get_stats();

datafusion/sqllogictest/test_files/file_row_index.slt

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -114,12 +114,26 @@ SELECT column1 FROM parquet_table WHERE file_row_index() > 2 ORDER BY column1
114114
statement ok
115115
RESET datafusion.execution.parquet.pushdown_filters;
116116

117-
# Without the rewrite in ParquetSource, `file_row_index()` is just NULL
117+
# Without the rewrite in ParquetSource, `file_row_index()` errors because it
118+
# depends on file-source context
118119

119-
query I
120+
query error file_row_index\(\) is source dependent and cannot be evaluated directly
120121
SELECT file_row_index()
121-
----
122-
NULL
122+
123+
# Testing pushdown over a source that doesn't support `file_row_index()`.
124+
125+
statement ok
126+
COPY (VALUES (10), (20), (30), (40), (50))
127+
TO 'test_files/scratch/file_row_index/csv_table/data.csv'
128+
STORED AS CSV;
129+
130+
statement ok
131+
CREATE EXTERNAL TABLE csv_table(column1 int)
132+
STORED AS CSV
133+
LOCATION 'test_files/scratch/file_row_index/csv_table/data.csv';
134+
135+
query error file_row_index\(\) is source dependent and cannot be evaluated directly
136+
SELECT *, file_row_index() FROM csv_table;
123137

124138
# Testing a table with two files.
125139

@@ -152,3 +166,6 @@ DROP TABLE parquet_two_files;
152166

153167
statement ok
154168
DROP TABLE parquet_table;
169+
170+
statement ok
171+
DROP TABLE csv_table;

docs/source/user-guide/sql/scalar_functions.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5894,8 +5894,8 @@ that produced the current row.
58945894
The value is scoped to one file, so rows from different files in the same scan
58955895
can have the same row index. This function is intended to be rewritten at
58965896
file-scan time. If the input file is not known (for example, if this function
5897-
is evaluated outside a file scan, or was not pushed down into one), this
5898-
function returns NULL.
5897+
is evaluated outside a file scan, or was not pushed down into one), direct
5898+
evaluation returns an error.
58995899

59005900
```sql
59015901
file_row_index()

0 commit comments

Comments
 (0)