feat: extend unnest to support Struct datatype

[duongcongtoai]

Which issue does this PR close?

Closes #10264.

Rationale for this change

What changes are included in this PR?

• Unnest logical plan is aware of the type of columns to be unnested
• UnnestExec (physical plan) expands the input batch both vertically and horizontally, depends on the set of columns to be unnested
• Modify try_process_unnest so it also interested in unnest(struct) expr, and translate this Expr into the projection of multiple columns for each subfield of the current struct in the column. Detailed expected behavior is in this test
• Add unnest(struct) scehanrios to unnest.slt, and one case where unnest is executed for different columns of type list/struct simultaneously

Are these changes tested?

Are there any user-facing changes?

[duongcongtoai]

I'm still finding a way to solve this TODO

[alamb]

When do we end up with an unest Expr (vs an LogicalPlan::Unnest).

Could we simply return not_yet_implemented error here?

[duongcongtoai]

if we returns error here, the error will stop the query plan from continue.
Initially unnest comes into projection step as an Expr here:

datafusion/datafusion/expr/src/utils.rs

Line 734 in d58bae4

exprs.iter().map(|e| e.to_field(input_schema)).collect()

, where it check data types of each expr, then later on we call
try_process_unnest to do the transformation from unnest(struct) -> struct.field1, struct.field 2 ...

I think unnest is the only expr so far that represent a set of fields instead of singular value here, so i'm thinking of adding an extra check at this function

datafusion/datafusion/expr/src/utils.rs

Line 734 in d58bae4

exprs.iter().map(|e| e.to_field(input_schema)).collect()

[alamb]

Thank you @duongcongtoai -- I think this is really cool.

I played around with this PR a bit locally and it works nicely. I have a few suggested tests and some simplification suggestions, but I also think those could be done as follow on PRs. Please let me know what you think

[alamb]

When do we end up with an unest Expr (vs an LogicalPlan::Unnest).

Could we simply return not_yet_implemented error here?

[alamb]

Could you please add some documentation explaining what this function does?

Also, I wonder if it needs to be pub or if we could leave it pub(crate) or not pub?

[alamb]

Can we please add some other tests:

1. Test the output type (e.g. select arrow_typeof(unnest(column1)))
2. Test nested structs (I tested it works well, but I think we should have some coverage)

Nested structs

> create or replace table t as values (struct('a', 'b', struct('c'))), (struct('d', 'e', struct('f')));
0 row(s) fetched.
Elapsed 0.031 seconds.

> select * from t;
+-----------------------------+
| column1                     |
+-----------------------------+
| {c0: a, c1: b, c2: {c0: c}} |
| {c0: d, c1: e, c2: {c0: f}} |
+-----------------------------+
2 row(s) fetched.
Elapsed 0.006 seconds.

> select unnest(column1) from t;
+----------------------+----------------------+----------------------+
| unnest(t.column1).c0 | unnest(t.column1).c1 | unnest(t.column1).c2 |
+----------------------+----------------------+----------------------+
| a                    | b                    | {c0: c}              |
| d                    | e                    | {c0: f}              |
+----------------------+----------------------+----------------------+
2 row(s) fetched.
Elapsed 0.013 seconds.

And

> create or replace table t as values (struct('a', 'b', [1,2,3])), (struct('x', 'y', [10,20]));
0 row(s) fetched.
Elapsed 0.010 seconds.

> select unnest(column1) from t;
+----------------------+----------------------+----------------------+
| unnest(t.column1).c0 | unnest(t.column1).c1 | unnest(t.column1).c2 |
+----------------------+----------------------+----------------------+
| a                    | b                    | [1, 2, 3]            |
| x                    | y                    | [10, 20]             |
+----------------------+----------------------+----------------------+
2 row(s) fetched.
Elapsed 0.008 seconds.

[alamb]

Interestingly the output type doesn't seem to work (which probably makes sense as unnest needs to be converted to a LogicalPlan). I don't think we have to fix this in the PR (but adding coverage would be good)

> select unnest(column1) from t;
+----------------------+----------------------+----------------------+
| unnest(t.column1).c0 | unnest(t.column1).c1 | unnest(t.column1).c2 |
+----------------------+----------------------+----------------------+
| a                    | b                    | [1, 2, 3]            |
| x                    | y                    | [10, 20]             |
+----------------------+----------------------+----------------------+
2 row(s) fetched.

> select arrow_typeof(unnest(column1)) from t;
Internal error: unnest on struct can ony be applied at the root level of select expression.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker

[duongcongtoai]

thank you, i added more coverage

[alamb]

I think you can make this a lot more concise using the expr_fn API, like:

Suggested change

	expr: Box::new(Expr::Column(Column::from_name("struct_col"))),
	expr: Box::new(col("struct_col"))

In fact maybe we should add an expr_fn for unnest

/// Call `unnest` on the expression
fn unnest(arg: Expr) -> Expr { 
  Expr::Unnest(Unnest { expr:: Box::new(arg) } )
}

[alamb]

Suggested change

	.add(Expr::Literal(ScalarValue::Int64(Some(1))));
	.add(lit(1i64));

There are similar simplifications we could do below too

[alamb]

Thanks again @duongcongtoai -- really nice improvement