Support ArrayType #536

elvaliuliuliu · 2020-06-05T08:42:06Z

This PR will add support to ArrayType which will fix part of #26 as follows:

Enable udf take in and return ArrayType by casting unpickled objects to the appropriate type. Support scenarios like simple array, array of arrays and array of Rows.
Convert internal SQL objects to c# objects when it's ArrayType.

laneser · 2020-06-25T23:00:42Z

This PR is good!
Because we have lots of parquet data which contains array type, and cannot read out by dotnet spark.
And this PR did the right job.

And, I saw the same check failed error that I cannot solve:

Test run for D:\a\1\s\master\artifacts\bin\Microsoft.Spark.Extensions.Delta.E2ETest\Release\netcoreapp3.1\Microsoft.Spark.Extensions.Delta.E2ETest.dll(.NETCoreApp,Version=v3.1)
Microsoft (R) Test Execution Command Line Tool Version 16.6.0
Copyright (c) Microsoft Corporation.  All rights reserved.

Starting test execution, please wait...

A total of 1 test files matched the specified pattern.
[2020-06-05T09:17:10.1955680Z] [fv-az352] [Info] [ConfigurationService] 'DOTNETBACKEND_PORT' environment variable is not set.
[2020-06-05T09:17:10.1970219Z] [fv-az352] [Info] [ConfigurationService] Using port 5567 for connection.
[2020-06-05T09:17:10.1973451Z] [fv-az352] [Info] [JvmBridge] JvMBridge port is 5567
[2020-06-05T09:18:00.8539213Z] [fv-az352] [Error] [JvmBridge] JVM method execution failed: Static method 'forPath' failed for class 'io.delta.tables.DeltaTable' when called with 1 arguments ([Index=1, Type=String, Value=C:\Users\VssAdministrator\AppData\Local\Temp\6cd7297f-60dd-463a-a915-b0ed79c2244f\sink-delta-table], )
[2020-06-05T09:18:00.8540601Z] [fv-az352] [Error] [JvmBridge] java.lang.IllegalArgumentException: Could not find active SparkSession

Maybe the E2E test sometimes just cannot find the active sparksession?

In my case, just re-run the test and it will passed...

suhsteve · 2020-06-25T23:04:58Z

This PR is good!
Because we have lots of parquet data which contains array type, and cannot read out by dotnet spark.
And this PR did the right job.

And, I saw the same check failed error that I cannot solve:

Test run for D:\a\1\s\master\artifacts\bin\Microsoft.Spark.Extensions.Delta.E2ETest\Release\netcoreapp3.1\Microsoft.Spark.Extensions.Delta.E2ETest.dll(.NETCoreApp,Version=v3.1)
Microsoft (R) Test Execution Command Line Tool Version 16.6.0
Copyright (c) Microsoft Corporation.  All rights reserved.

Starting test execution, please wait...

A total of 1 test files matched the specified pattern.
[2020-06-05T09:17:10.1955680Z] [fv-az352] [Info] [ConfigurationService] 'DOTNETBACKEND_PORT' environment variable is not set.
[2020-06-05T09:17:10.1970219Z] [fv-az352] [Info] [ConfigurationService] Using port 5567 for connection.
[2020-06-05T09:17:10.1973451Z] [fv-az352] [Info] [JvmBridge] JvMBridge port is 5567
[2020-06-05T09:18:00.8539213Z] [fv-az352] [Error] [JvmBridge] JVM method execution failed: Static method 'forPath' failed for class 'io.delta.tables.DeltaTable' when called with 1 arguments ([Index=1, Type=String, Value=C:\Users\VssAdministrator\AppData\Local\Temp\6cd7297f-60dd-463a-a915-b0ed79c2244f\sink-delta-table], )
[2020-06-05T09:18:00.8540601Z] [fv-az352] [Error] [JvmBridge] java.lang.IllegalArgumentException: Could not find active SparkSession

Maybe the E2E test sometimes just cannot find the active sparksession?

In my case, just re-run the test and it will passed...

@laneser this error is related to #333 . Internally Delta will call getActiveSession() and if it's lucky then it will use the thread where the active session is defined.

laneser · 2020-06-26T03:47:56Z

@suhsteve
Do you mean that just change the

DeltaTable deltaTable = DeltaTable.ForPath(path);

To

DeltaTable deltaTable = DeltaTable.ForPath(_spark, path);

spark/src/csharp/Extensions/Microsoft.Spark.Extensions.Delta.E2ETest/DeltaTableTests.cs

Line 51 in ce23177

DeltaTable deltaTable = DeltaTable.ForPath(path);

will fix the unit-test ?! which do not call ForPath (use GetActiveSession version)?

elvaliuliuliu · 2020-06-26T03:55:59Z

I will try get this PR in soon!

imback82 · 2020-06-27T17:21:21Z

Thanks @elvaliuliuliu. What work is remaining to make this non-WIP?

elvaliuliuliu · 2020-06-27T17:28:37Z

Thanks @elvaliuliuliu. What work is remaining to make this non-WIP?

Currently, udf takes in ArrayList like the example here, I am thinking to support this. Should I include it in this PR?

imback82 · 2020-06-27T17:42:41Z

Got it. Yes, let's include it in this PR. Thanks.

rapoth · 2020-08-10T23:58:09Z

@elvaliuliuliu Were you planning on updating this PR?

elvaliuliuliu · 2020-08-11T00:05:45Z

Sorry got side-tracked. I will update this PR!

elvaliuliuliu · 2020-08-19T22:30:51Z

This PR should be ready for review. I think FC and BC tests failed as expected, it should work with the current code, please advise. Thank you!

Niharikadutta · 2020-08-19T23:29:20Z

Thanks for working on this @elvaliuliuliu !
@imback82 @suhsteve Do we want to exclude these tests from the FC and BC pipelines?

src/csharp/Microsoft.Spark/Utils/PythonSerDe.cs

imback82 · 2020-08-20T17:03:35Z

src/csharp/Microsoft.Spark/Utils/PythonSerDe.cs

+                        }
+                        if (obj.GetType() == typeof(ArrayList))
+                        {
+                            return CastUnpickledItems.UnpickleArray(unpickledItems);


if you just return here, what will happen to the rest of rows? Would that be ok?

I think it should be okay, since UnpickleArray takes unpickledItems and within the func itself, it will deal with all the rows when it's ArrayList.

If you get ArrayList, all the objects in unpickledItems will be a type of ArrayList?

Yes, from what I have observed. It will look like below:

Maybe we should change this function to IEnumerable<object> GetUnpickledObjects and use yield return.

Do you mean something like below?

internal static IEnumerable<object> GetUnpickledObjects(Stream stream, int messageLength) { byte[] buffer = ArrayPool<byte>.Shared.Rent(messageLength); try { ... // Check if unpickler returns ArrayList. // If so, it needs to be cast to the appropriate array type using CastToArray. foreach (object objArr in (object[])unpickledItems) { if (objArr.GetType() == typeof(object[])) { object obj = ((object[])objArr)[0]; if (obj == null) { continue; } if (obj.GetType() == typeof(ArrayList)) { yield return CastUnpickledItems.CastToArray(unpickledItems); } else { yield return unpickledItems; } } else { yield return unpickledItems; } } } finally { ArrayPool<byte>.Shared.Return(buffer); } }

I agree with @imback82 . You are iterating through all the object[] entries here, and the consumer of GetUnpickledObjects will also end up iterating through the object[] again. It may be better to make this an IEnumerable

Sure, will make the change tonight.

I have refactored this part. Please take a look, thanks!

… into elva/arrayType

imback82 · 2020-08-24T23:09:05Z

src/csharp/Microsoft.Spark/Utils/PythonSerDe.cs

+                        }
+                        if (obj.GetType() == typeof(ArrayList))
+                        {
+                            return CastUnpickledItems.UnpickleArray(unpickledItems);


Maybe we should change this function to IEnumerable<object> GetUnpickledObjects and use yield return.

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs

src/csharp/Microsoft.Spark/Sql/Types/ComplexTypes.cs

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs

suhsteve · 2020-09-16T00:25:24Z

@elvaliuliuliu how's the progress going? Tests are failing.

elvaliuliuliu · 2020-09-16T00:35:59Z

@elvaliuliuliu how's the progress going? Tests are failing.

Working on n d arrays. Some of the tests are passing, it failed at certain spark versions with error like "System.IO.DirectoryNotFoundException : Could not find a part of the path 'D:\a\1\b\spark-2.3.4-bin-hadoop2.7\RELEASE'", looking into it.

src/csharp/Microsoft.Spark/Sql/Types/SimpleTypes.cs

suhsteve · 2020-09-16T17:11:16Z

src/csharp/Microsoft.Spark.E2ETest/UdfTests/UdfComplexTypesTests.cs

+
+            // Array of Arrays.
+            {
+                Func<Column, Column> udf = Udf<double[][], double>(


can you pass in multiple arrays ?

Udf<double, double[], double[][], double[][][], double> ?

Can we give the user the option to also define a udf using ArrayList? Something like
Udf<double, ArrayList, ArrayList, ArrayList, double> that will have the same behavior as the udf defined above ?

Added tests for multiple arrays. Working on ArrayList cases as mentioned.

Do we also want to support the option of return value as ArrayList?
Like Udf<double, ArrayList, ... , ArrayList>?

@imback82 ideas?

I guess it can be useful if you want to chain with Udf that takes in ArrayList (just for consistency).

Got it, thanks!

The current logic is to cast unpickledItems to typed array if it contains ArrayList here. Another way might be casting input(after unpickling) here when executing udf functions?

I am thinking either to pass commands in GetUnpickledObjects or switch to the latter method, if we want to support ArrayList like Udf<ArrayList, ArrayList, double>. Please lmk which is preferred or any other suggestions? Thanks.

suhsteve · 2020-09-17T17:14:33Z

We are targeting a release by end of month and we would like this PR in within the next 2-3 days.

Can we get everything wrapped up by today ?

elvaliuliuliu · 2020-09-17T18:10:37Z

We are targeting a release by end of month and we would like this PR in within the next 2-3 days.

Can we get everything wrapped up by today ?

I am finishing up the following left parts now - Updated

Support ArrayList when defining udf. Need inputs/ideas.
Change to IEnumerable<object> GetUnpickledObjects.
Clean up and other changes. (CastToRowArray(), GetUnpickledObjects, 2 comments for simpletypes).

Please lmk if more comments. Thanks.

imback82 · 2020-09-18T20:32:54Z

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs

+            foreach (object obj in (object[])unpickledItems)
+            {
+                castUnpickledItems.Add(
+                    (obj.GetType() == typeof(RowConstructor)) ? obj : CastArray(obj));


I am not sure if falling back to CastArray always works.

I have updated the function name accordingly. Sorry for the confusion. If not RowConstructor, it will fall into object []. CastHelper will help decide and cast object[] as needed.

imback82 · 2020-09-18T20:33:07Z

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs

+        /// <returns>Typed array after casting.</returns>
+        public static object CastArray(object obj)
+        {
+            if (obj is object[] objArr)


does this work with Row[]?

From what I have observed, if udf takes Row[] like the example here. It will be RowConstructor, and FromInternal should handle such cases. It should not fall into CastArray or CastHelper.

imback82 · 2020-09-18T20:33:34Z

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs

+                return objArr.Select(x => CastHelper(x)).ToArray();
+            }
+
+            // Array of arrays.


how about array of array of array? Don't you need to handle this recursively?

I have changed the function name and description accordingly to make it more clear. This should be handled recursively as covered in test. Thanks!

… into elva/arrayType

suhsteve · 2020-09-24T20:15:07Z

Closing as #670 has been merged.

elvaliuliuliu and others added 3 commits June 5, 2020 01:34

init

ece2b69

clean up

67e9e46

Merge branch 'master' into elva/arrayType

7e0b5b4

elvaliuliuliu mentioned this pull request Jun 12, 2020

dotnet-interactive assembly extension #517

Merged

Merge branch 'master' into elva/arrayType

69b5c0b

Merge branch 'master' into elva/arrayType

e3f4009

Niharikadutta mentioned this pull request Aug 12, 2020

Exposing DataFrame.Persist(StorageLevel) and DataFrame.StorageLevel APIs #623

Merged

elvaliuliuliu and others added 2 commits August 17, 2020 20:24

cast unpickled items

0dfdacd

Merge branch 'master' into elva/arrayType

1f218db

elvaliuliuliu changed the title ~~[WIP] Support ArrayType~~ Support ArrayType Aug 18, 2020

imback82 reviewed Aug 20, 2020

View reviewed changes

elvaliuliuliu and others added 4 commits August 24, 2020 13:03

Merge branch 'master' into elva/arrayType

25bced0

PR comments

09e5b51

Merge branch 'elva/arrayType' of https://github.com/elvaliuliuliu/spark…

b262c22

… into elva/arrayType

add more comments

e162ff8

imback82 reviewed Aug 24, 2020

View reviewed changes

elvaliuliuliu and others added 2 commits August 25, 2020 14:40

change func name to cast

dcdf49d

Merge branch 'master' into elva/arrayType

b8a09c6

Niharikadutta reviewed Aug 27, 2020

View reviewed changes

src/csharp/Microsoft.Spark/Sql/CastUnpickledItems.cs Outdated Show resolved Hide resolved

elvaliuliuliu added 3 commits September 14, 2020 17:56

clean up

19a8218

fix null

547f43a

PR reviews

dd80dca

add nd array support

519de85

suhsteve reviewed Sep 16, 2020

View reviewed changes

elvaliuliuliu added 4 commits September 16, 2020 15:31

update array of arrays test

7e2aa92

clean up

f1d3f04

nit

3cb5210

retrigger pipeline for active session bug

13fff5d

elvaliuliuliu added 5 commits September 17, 2020 20:22

wrap up

65b4013

add description

bdce181

PR reviews & clean up

1c47d41

fix unit test

01cd49b

update

893d785

imback82 reviewed Sep 18, 2020

View reviewed changes

elvaliuliuliu and others added 9 commits September 18, 2020 14:11

update func - PR review

36c47bf

Merge branch 'master' into elva/arrayType

25abbfa

nit

42c5dde

Merge branch 'elva/arrayType' of https://github.com/elvaliuliuliu/spark…

ad1c588

… into elva/arrayType

nit

1d3a2c4

clean up

aeed975

nit & retrigger pipeline for active session error

6402870

clean up

dac2016

revert

ad06296

suhsteve closed this Sep 24, 2020

Support ArrayType #536

Support ArrayType #536

Uh oh!

Conversation

elvaliuliuliu commented Jun 5, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

laneser commented Jun 25, 2020

Uh oh!

suhsteve commented Jun 25, 2020

Uh oh!

laneser commented Jun 26, 2020

Uh oh!

elvaliuliuliu commented Jun 26, 2020

Uh oh!

imback82 commented Jun 27, 2020

Uh oh!

elvaliuliuliu commented Jun 27, 2020

Uh oh!

imback82 commented Jun 27, 2020

Uh oh!

rapoth commented Aug 10, 2020

Uh oh!

elvaliuliuliu commented Aug 11, 2020

Uh oh!

elvaliuliuliu commented Aug 19, 2020

Uh oh!

Niharikadutta commented Aug 19, 2020

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

suhsteve commented Sep 16, 2020

Uh oh!

elvaliuliuliu commented Sep 16, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

suhsteve Sep 16, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

elvaliuliuliu Sep 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

suhsteve commented Sep 17, 2020

Uh oh!

elvaliuliuliu commented Sep 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

elvaliuliuliu commented Jun 5, 2020 •

edited

Loading

elvaliuliuliu commented Sep 16, 2020 •

edited

Loading

suhsteve Sep 16, 2020 •

edited

Loading

elvaliuliuliu Sep 17, 2020 •

edited

Loading

elvaliuliuliu commented Sep 17, 2020 •

edited

Loading

elvaliuliuliu Sep 18, 2020 •

edited

Loading