fix: propagate errors for unsupported table function arguments instead of silently dropping them (#21135)

## Which issue does this PR close?
- Closes #21125.

## Rationale for this change
`flat_map` swallows error since it results into an iterator that yields
no elements. Instead of this this PR uses `map` and propagates error.

## What changes are included in this PR?
error propagation and related tests

## Are these changes tested?
Yes I've added both unit and slt test

## Are there any user-facing changes?
User's will see propagated error

Author: buraksenn

datafusion/sql/src/relation/mod.rs

@@ -151,7 +151,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                     let args = func_args
                         .args
                         .into_iter()
-                        .flat_map(|arg| {
+                        .map(|arg| {
                             if let FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) = arg
                             {
                                 self.sql_expr_to_logical_expr(
@@ -163,7 +163,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                                 plan_err!("Unsupported function argument type: {}", arg)
                             }
                         })
-                        .collect::<Vec<_>>();
+                        .collect::<Result<Vec<_>>>()?;
                     let provider = self
                         .context_provider
                         .get_table_function_source(&tbl_func_name, args)?;

datafusion/sql/tests/sql_integration.rs

@@ -4725,6 +4725,26 @@ fn plan_create_index() {
     }
 }
 
+#[test]
+fn test_table_function_with_unsupported_arg_propagates_error() {
+    let sql = "SELECT * FROM my_func(('a', 'b', 'c'))";
+    let dialect = &GenericDialect {};
+    let state = MockSessionState::default();
+    let context = MockContextProvider { state };
+    let planner = SqlToRel::new(&context);
+    let result = DFParser::parse_sql_with_dialect(sql, dialect);
+    let mut ast = result.unwrap();
+    let err = planner
+        .statement_to_plan(ast.pop_front().unwrap())
+        .expect_err("query should have failed");
+    let msg = err.strip_backtrace();
+    assert!(
+        !msg.contains("Table Functions are not supported"),
+        "tuple argument error should be propagated before reaching get_table_function_source, got: {msg}"
+    );
+    assert_contains!(msg, "Struct not supported");
+}
+
 fn assert_field_not_found(mut err: DataFusionError, name: &str) {
     let err = loop {
         match err {

datafusion/sqllogictest/test_files/table_functions.slt

@@ -397,7 +397,7 @@ SELECT * FROM range(TIMESTAMP '2023-01-03T00:00:00', TIMESTAMP '2023-01-01T00:00
 
 # Single timestamp (start == end)
 query P
-SELECT * FROM range(TIMESTAMP '2023-01-01T00:00:00', TIMESTAMP '2023-01-01T00:00:00', INTERVAL '1' DAY)  
+SELECT * FROM range(TIMESTAMP '2023-01-01T00:00:00', TIMESTAMP '2023-01-01T00:00:00', INTERVAL '1' DAY)
 ----
 
 # Timestamp range with NULL values
@@ -543,3 +543,16 @@ LIMIT 10;
 1
 1
 1
+
+# Test that unsupported function argument types are properly reported
+# rather than being silently dropped (which previously caused a misleading
+# "requires 1 to 3 arguments" error instead)
+
+statement error DataFusion error: Error during planning: Unsupported function argument type: start => 1
+SELECT * FROM generate_series(start => 1, stop => 5)
+
+statement error DataFusion error: Error during planning: Unsupported function argument type: \*
+SELECT * FROM generate_series(*)
+
+statement error DataFusion error: Error during planning: Unsupported function argument type: t\.\*
+SELECT * FROM generate_series(t.*)