[SPARK-29378][R] Upgrade SparkR to use Arrow 0.15 API

[dongjoon-hyun]

What changes were proposed in this pull request?

[SPARK-29376] Upgrade Apache Arrow to version 0.15.1 upgrades to Arrow 0.15 at Scala/Java/Python. This PR aims to upgrade SparkR to use Arrow 0.15 API. Currently, it's broken.

Why are the changes needed?

First of all, it turns out that our Jenkins jobs (including PR builder) ignores Arrow test. Arrow 0.15 has a breaking R API changes at ARROW-5505 and we missed that. AppVeyor was the only one having SparkR Arrow tests but it's broken now.

Jenkins

Skipped ------------------------------------------------------------------------
1. createDataFrame/collect Arrow optimization (@test_sparkSQL_arrow.R#25)
- arrow not installed

Second, Arrow throws OOM on AppVeyor environment (Windows JDK8) like the following because it still has Arrow 0.14.

Warnings -----------------------------------------------------------------------
1. createDataFrame/collect Arrow optimization (@test_sparkSQL_arrow.R#39) - createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.sparkr.enabled' is set to true; however, failed, attempting non-optimization. Reason: Error in handleErrors(returnStatus, conn): java.lang.OutOfMemoryError: Java heap space
	at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
	at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
	at org.apache.arrow.vector.ipc.message.MessageSerializer.readMessage(MessageSerializer.java:669)
	at org.apache.spark.sql.execution.arrow.ArrowConverters$$anon$3.readNextBatch(ArrowConverters.scala:243)

It is due to the version mismatch.

int messageLength = MessageSerializer.bytesToInt(buffer.array());
if (messageLength == IPC_CONTINUATION_TOKEN) {
  buffer.clear();
  // ARROW-6313, if the first 4 bytes are continuation message, read the next 4 for the length
  if (in.readFully(buffer) == 4) {
    messageLength = MessageSerializer.bytesToInt(buffer.array());
  }
}

// Length of 0 indicates end of stream
if (messageLength != 0) {
  // Read the message into the buffer.
  ByteBuffer messageBuffer = ByteBuffer.allocate(messageLength);

After upgrading this to 0.15, we are hitting ARROW-5505. This PR upgrades Arrow version in AppVeyor and fix the issue.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Pass the AppVeyor.

This PR passed here.

• https://ci.appveyor.com/project/ApacheSoftwareFoundation/spark/builds/28909044

SparkSQL Arrow optimization: Spark package found in SPARK_HOME: C:\projects\spark\bin\..
................

[dongjoon-hyun]

AppVeyor is broken as we see in this PR.

• https://ci.appveyor.com/project/ApacheSoftwareFoundation/spark/builds/28902071

[srowen]

It seems like a good idea to match these, yeah. Different error now?

6. Error: gapply() Arrow optimization - type specification (date and timestamp) (@test_sparkSQL_arrow.R#285) 
invalid 'n' argument
1: tryCatch({
       ret <- gapply(df, "a", function(key, grouped) {
           grouped
       }, schema(df))
       expect_equal(collect(ret), rdf)
   }, finally = {
       callJMethod(conf, "set", "spark.sql.execution.arrow.sparkr.enabled", arrowEnabled)
   }) at C:\projects\spark\bin\../R/pkg/tests/fulltests/test_sparkSQL_arrow.R:285
2: tryCatchList(expr, classes, parentenv, handlers)
3: expect_equal(collect(ret), rdf) at C:\projects\spark\bin\../R/pkg/tests/fulltests/test_sparkSQL_arrow.R:289
4: compare(object, expected, ...)
5: collect(ret)
6: collect(ret)
7: .local(x, ...)
8: tryCatch({
       doServerAuth(conn, authSecret)
       arrowTable <- read_arrow(readRaw(conn))
       if (useAsTibble) {
           as_tibble <- get("as_tibble", envir = asNamespace("arrow"))
           as.data.frame(as_tibble(arrowTable), stringsAsFactors = stringsAsFactors)
       }
       else {
           as.data.frame(arrowTable, stringsAsFactors = stringsAsFactors)
       }
   }, finally = {
       close(conn)
   })
9: tryCatchList(expr, classes, parentenv, handlers)
10: read_arrow(readRaw(conn))
11: as.data.frame(read_table(stream))
12: read_table(stream)
13: readRaw(conn)
14: readBin(con, raw(), as.integer(dataLen), endian = "big")

Still seems like some protocol / version mismatch. I don't know much about this but looks like readRaw gets back info from conn that causes it to try to read an invalid amount, like -1. I think that kind of "-1" return has something to do with being out of memory, so maybe back where we started.

Does the machine need more memory or is that irrelevant?

[dongjoon-hyun]

Correct. The previous one was due to version mismatch. messageLength seems to be read in a wrong way.

      int messageLength = MessageSerializer.bytesToInt(buffer.array());
      if (messageLength == IPC_CONTINUATION_TOKEN) {
        buffer.clear();
        // ARROW-6313, if the first 4 bytes are continuation message, read the next 4 for the length
        if (in.readFully(buffer) == 4) {
          messageLength = MessageSerializer.bytesToInt(buffer.array());
        }
      }

      // Length of 0 indicates end of stream
      if (messageLength != 0) {

        // Read the message into the buffer.
        ByteBuffer messageBuffer = ByteBuffer.allocate(messageLength);

And, yes. Now, with this PR, it seems to show the previous blocker of SPARK-29378?

1. createDataFrame/collect Arrow optimization (@test_sparkSQL_arrow.R#39) - createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.sparkr.enabled' is set to true; however, failed, attempting non-optimization.
Reason: Error in arrow::FileOutputStream(fileName): attempt to apply non-function

[dongjoon-hyun]

ARROW-5505 changes their R API. cc @srowen , @HyukjinKwon , @felixcheung

[dongjoon-hyun]

@shaneknapp . It seems that our Jenkins doesn't have R Arrow installation and skip these test.

[dongjoon-hyun]

This is not Windows/JDK8 issue. Arrow 0.15 has a breaking change in R interface.

[shaneknapp]

@shaneknapp . It seems that our Jenkins doesn't have R Arrow installation and skip these test.

this is correct. should i infer that we need to add R Arrow to our worker nodes?

[HyukjinKwon]

Actually, I configured AppVeyor to test Arrow related ones in R for now. So, at least it's being tested in AppVeyor.

Yes, it would be nice if we have Arrow R installed in Jenkins workers to test it out; however, I remember it's a bit tricky to install Arrow R library for now (you cannot just install via CRAN but needs some manual preparation - see https://github.com/apache/arrow/tree/master/r#installation).

I am worried if it messes up the current Jenkins worker's environment. If you see any concern, @shaneknapp, you can just don't install it for now. It's being tested via AppVeyor at least so it's neither a must or urgent.

[shaneknapp]

actually, R arrow installed easily and w/o issue (surprising, i know!). :)

sknapp@ubuntu-testing:~$ Rscript -e "packageVersion('arrow')"
[1] ‘0.15.1.1’

i will install this on the remaining workers later today/tomorrow.

[dongjoon-hyun]

Thank you, @shaneknapp !

[HyukjinKwon]

Thank you!

[BryanCutler]

Thanks for fixing this @dongjoon-hyun !

[BryanCutler]

Would it be better to restore this line from #26041 to install arrow?

cmd: R -e "install.packages(c('knitr', 'rmarkdown', 'e1071', 'survival', 'arrow'), repos='https://cloud.r-project.org/')"

[dongjoon-hyun]

Thanks. Yes. That will be better.

[dongjoon-hyun]

For the other errors, I'm working on.

[SparkQA]

Test build #113945 has finished for PR 26555 at commit 96f20f8.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

Now, this PR passed all R tests (Jenkins + AppVeyor including Arrow 0.15).
Could you review this PR again, @srowen , @BryanCutler , @HyukjinKwon , @felixcheung , @gatorsmile ?

[viirya]

I think this means the minimum version is 0.15? Then we should also update document like sparkr.md.

[dongjoon-hyun]

There is already a documentation JIRA for that, @viirya

• https://issues.apache.org/jira/browse/SPARK-29924

[viirya]

@dongjoon-hyun Thanks! I see. Then let's update the R document in that.

[viirya]

From the Arrow PR apache/arrow#5279, this change looks good.

[dongjoon-hyun]

Thank you all for review. I'll merge this PR to recover the AppVeyor build and test.
Since it's only for Window-environment, we need to enable this arrow test in Jenkins to protect our branch.

[HyukjinKwon]

Havnt taken a close look but looks good. Thanks @dongjoon-hyun

[dongjoon-hyun]

Thanks, @HyukjinKwon !

[HyukjinKwon]

This line was to install Arrow dependencies manually (because devtools started to require testthat dependency as 2.0.0+; however, SparkR requires 1.0.2). So, I think we can remove this line too. Let me make a quick followup.

[dongjoon-hyun]

Thanks. Go ahead!

[HyukjinKwon]

Made a followup - #26566. For clarification It actually makes no diff but only less codes.