Skip to content

Fix the chained UDF with Row type #411

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 17 commits into from
Feb 8, 2020
Merged

Fix the chained UDF with Row type #411

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
274f7e1
init
elvaliuliuliu Feb 1, 2020
1fa97ff
fix
elvaliuliuliu Feb 1, 2020
9828a99
resolve comments
elvaliuliuliu Feb 4, 2020
87caf70
add GenericRow to Row conversions
elvaliuliuliu Feb 4, 2020
2110e0b
fix
elvaliuliuliu Feb 4, 2020
728f406
clean up
elvaliuliuliu Feb 4, 2020
48f92f4
change to implicit
elvaliuliuliu Feb 4, 2020
7e3dbc5
fix
elvaliuliuliu Feb 4, 2020
57574ab
clean up
elvaliuliuliu Feb 4, 2020
e829d78
add GenericRowPickler for nested Rows
elvaliuliuliu Feb 4, 2020
9938ca9
remove spaces
elvaliuliuliu Feb 4, 2020
44f5635
resolve comments
elvaliuliuliu Feb 4, 2020
290df45
fixes
elvaliuliuliu Feb 4, 2020
db0eba8
Merge branch 'master' into elva/chainedUdf
imback82 Feb 6, 2020
35e6441
resolve comments
elvaliuliuliu Feb 7, 2020
ebdccec
Merge branch 'elva/chainedUdf' of https://github.com/elvaliuliuliu/sp…
elvaliuliuliu Feb 7, 2020
de803d1
fix
elvaliuliuliu Feb 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 33 additions & 12 deletions src/csharp/Microsoft.Spark.E2ETest/UdfTests/UdfComplexTypesTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ public void TestUdfWithReturnAsMapType()
[Fact]
public void TestUdfWithRowType()
{
// Single Row
// Single Row.
{
Func<Column, Column> udf = Udf<Row, string>(
(row) => row.GetAs<string>("city"));
Expand All @@ -149,7 +149,7 @@ public void TestUdfWithRowType()
Assert.Equal(expected, actual);
}

// Multiple Rows
// Multiple Rows.
{
Func<Column, Column, Column, Column> udf = Udf<Row, Row, string, string>(
(row1, row2, str) =>
Expand All @@ -173,7 +173,7 @@ public void TestUdfWithRowType()
Assert.Equal(expected, actual);
}

// Nested Row
// Nested Rows.
{
Func<Column, Column> udf = Udf<Row, string>(
(row) =>
Expand All @@ -197,15 +197,15 @@ public void TestUdfWithRowType()
[Fact]
public void TestUdfWithReturnAsRowType()
{
// Single GenericRow
// Single Row.
{
var schema = new StructType(new[]
{
new StructField("col1", new IntegerType()),
new StructField("col2", new StringType())
});
Func<Column, Column> udf = Udf<string>(
str => new GenericRow(new object[] { 1, "abc" }), schema);
str => new Row(new object[] { 1, "abc" }), schema);

Row[] rows = _df.Select(udf(_df["name"]).As("col")).Collect().ToArray();
Assert.Equal(3, rows.Length);
Expand All @@ -219,14 +219,14 @@ public void TestUdfWithReturnAsRowType()
}
}

// Generic row is a part of top-level column.
// Row is a part of top-level column.
{
var schema = new StructType(new[]
{
new StructField("col1", new IntegerType())
});
Func<Column, Column> udf = Udf<string>(
str => new GenericRow(new object[] { 111 }), schema);
str => new Row(new object[] { 111 }), schema);

Column nameCol = _df["name"];
Row[] rows = _df.Select(udf(nameCol).As("col"), nameCol).Collect().ToArray();
Expand All @@ -244,7 +244,7 @@ public void TestUdfWithReturnAsRowType()
}
}

// Nested GenericRow
// Nested Rows.
{
var subSchema1 = new StructType(new[]
{
Expand All @@ -263,15 +263,15 @@ public void TestUdfWithReturnAsRowType()
});

Func<Column, Column> udf = Udf<string>(
str => new GenericRow(
str => new Row(
new object[]
{
1,
new GenericRow(new object[] { 1 }),
new GenericRow(new object[]
new Row(new object[] { 1 }),
new Row(new object[]
{
"abc",
new GenericRow(new object[] { 10 })
new Row(new object[] { 10 })
})
}),
schema);
Expand All @@ -295,6 +295,27 @@ public void TestUdfWithReturnAsRowType()
outerCol.GetAs<Row>("col3"));
}
}

// Chained UDFs with Row type.
{
var schema = new StructType(new[]
{
new StructField("col1", new IntegerType()),
new StructField("col2", new StringType())
});
Func<Column, Column> udf1 = Udf<string>(
str => new Row(new object[] { 1, "abc" }), schema);

Func<Column, Column> udf2 = Udf<Row, string>(
row => row.GetAs<string>(1));

Row[] rows = _df.Select(udf2(udf1(_df["name"]))).Collect().ToArray();
Assert.Equal(3, rows.Length);

var expected = new[] { "abc", "abc", "abc" };
string[] actual = rows.Select(x => x[0].ToString()).ToArray();
Assert.Equal(expected, actual);
}
}
}
}
2 changes: 1 addition & 1 deletion src/csharp/Microsoft.Spark.UnitTest/Sql/RowTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ public void RowCollectorTest()
private Pickler CreatePickler()
{
new StructTypePickler().Register();
new RowPickler().Register();
new Microsoft.Spark.UnitTest.TestUtils.RowPickler().Register();
return new Pickler();
}

Expand Down
22 changes: 11 additions & 11 deletions src/csharp/Microsoft.Spark/Sql/Functions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3804,7 +3804,7 @@ public static Func<Column, Column, Column, Column, Column, Column, Column, Colum
/// <returns>
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column> Udf(Func<GenericRow> udf, StructType returnType)
public static Func<Column> Udf(Func<Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply0;
}
Expand All @@ -3816,7 +3816,7 @@ public static Func<Column> Udf(Func<GenericRow> udf, StructType returnType)
/// <returns>
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column> Udf<T>(Func<T, GenericRow> udf, StructType returnType)
public static Func<Column, Column> Udf<T>(Func<T, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply1;
}
Expand All @@ -3830,7 +3830,7 @@ public static Func<Column, Column> Udf<T>(Func<T, GenericRow> udf, StructType re
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column> Udf<T1, T2>(
Func<T1, T2, GenericRow> udf, StructType returnType)
Func<T1, T2, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply2;
}
Expand All @@ -3845,7 +3845,7 @@ public static Func<Column, Column, Column> Udf<T1, T2>(
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column> Udf<T1, T2, T3>(
Func<T1, T2, T3, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply3;
}
Expand All @@ -3861,7 +3861,7 @@ public static Func<Column, Column, Column, Column> Udf<T1, T2, T3>(
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4>(
Func<T1, T2, T3, T4, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply4;
}
Expand All @@ -3878,7 +3878,7 @@ public static Func<Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4>(
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5>(
Func<T1, T2, T3, T4, T5, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply5;
}
Expand All @@ -3896,7 +3896,7 @@ public static Func<Column, Column, Column, Column, Column, Column> Udf<T1, T2, T
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5, T6>(
Func<T1, T2, T3, T4, T5, T6, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, T6, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply6;
}
Expand All @@ -3915,7 +3915,7 @@ public static Func<Column, Column, Column, Column, Column, Column, Column> Udf<T
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5, T6, T7>(
Func<T1, T2, T3, T4, T5, T6, T7, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, T6, T7, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply7;
}
Expand All @@ -3935,7 +3935,7 @@ public static Func<Column, Column, Column, Column, Column, Column, Column, Colum
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5, T6, T7, T8>(
Func<T1, T2, T3, T4, T5, T6, T7, T8, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, T6, T7, T8, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply8;
}
Expand All @@ -3954,7 +3954,7 @@ public static Func<Column, Column, Column, Column, Column, Column, Column, Colum
/// <param name="returnType">Schema associated with this row</param>
/// <returns>A delegate that when invoked will return a <see cref="Column"/> for the result of the UDF.</returns>
public static Func<Column, Column, Column, Column, Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5, T6, T7, T8, T9>(
Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply9;
}
Expand All @@ -3976,7 +3976,7 @@ public static Func<Column, Column, Column, Column, Column, Column, Column, Colum
/// A delegate that returns a <see cref="Column"/> for the result of the UDF.
/// </returns>
public static Func<Column, Column, Column, Column, Column, Column, Column, Column, Column, Column, Column> Udf<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>(
Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, GenericRow> udf, StructType returnType)
Func<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, Row> udf, StructType returnType)
{
return CreateUdf(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply10;
}
Expand Down
9 changes: 9 additions & 0 deletions src/csharp/Microsoft.Spark/Sql/Row.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,15 @@ internal Row(object[] values, StructType schema)
Convert();
}

/// <summary>
/// Constructor for the Row class.
/// </summary>
/// <param name="values">Column values for a row</param>
internal Row(object[] values)
{
_genericRow = new GenericRow(values);
}

/// <summary>
/// Schema associated with this row.
/// </summary>
Expand Down
4 changes: 2 additions & 2 deletions .../Microsoft.Spark/Sql/GenericRowPickler.cs → src/csharp/Microsoft.Spark/Sql/RowPickler.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,11 @@ namespace Microsoft.Spark.Sql
/// <summary>
/// Custom pickler for GenericRow objects.
/// </summary>
internal class GenericRowPickler : IObjectPickler
internal class RowPickler : IObjectPickler
{
public void pickle(object o, Stream outs, Pickler currentPickler)
{
currentPickler.save(((GenericRow)o).Values);
currentPickler.save(((Row)o).Values);
}
}
}
4 changes: 2 additions & 2 deletions src/csharp/Microsoft.Spark/Utils/PythonSerDe.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,8 @@ static PythonSerDe()
Unpickler.registerConstructor(
"pyspark.sql.types", "_create_row_inbound_converter", s_rowConstructor);

// Register custom pickler for GenericRow objects.
Pickler.registerCustomPickler(typeof(GenericRow), new GenericRowPickler());
// Register custom pickler for Row objects.
Pickler.registerCustomPickler(typeof(Row), new RowPickler());
}

/// <summary>
Expand Down