[NSE-728] Upgrade to Arrow 7.0.0 (#729)

Known issues of current Arrow 7.0.0 support: 1. Data Source writing / ORC reading is disabled; 2. Data Source filter pushdown is disabled; 3. FastPFor compression is leading to unexpected concurrent memory writes. Use LZ4 instead.
oap-project · Apr 13, 2022 · e329253 · e329253
1 parent 8c181af
commit e329253
Show file tree

Hide file tree

Showing 30 changed files with 244 additions and 342 deletions.
diff --git a/.github/workflows/tpch.yml b/.github/workflows/tpch.yml
@@ -51,7 +51,7 @@ jobs:
         run: |
           cd /tmp
           git clone https://github.com/oap-project/arrow.git
-          cd arrow && git checkout arrow-4.0.0-oap && cd cpp
+          cd arrow && git checkout arrow-7.0.0-oap && cd cpp
           mkdir build && cd build
           cmake .. -DARROW_JNI=ON -DARROW_GANDIVA_JAVA=ON -DARROW_GANDIVA=ON -DARROW_PARQUET=ON -DARROW_ORC=ON -DARROW_CSV=ON -DARROW_HDFS=ON -DARROW_FILESYSTEM=ON -DARROW_WITH_SNAPPY=ON -DARROW_JSON=ON -DARROW_DATASET=ON -DARROW_WITH_LZ4=ON -DARROW_JEMALLOC=OFF && make -j2
           sudo make install

diff --git a/.github/workflows/unittests.yml b/.github/workflows/unittests.yml
@@ -45,7 +45,7 @@ jobs:
         run: |
           cd /tmp
           git clone https://github.com/oap-project/arrow.git
-          cd arrow && git checkout arrow-4.0.0-oap && cd cpp
+          cd arrow && git checkout arrow-7.0.0-oap && cd cpp
           mkdir build && cd build
           cmake .. -DARROW_JNI=ON -DARROW_GANDIVA_JAVA=ON -DARROW_GANDIVA=ON -DARROW_PARQUET=ON -DARROW_ORC=ON -DARROW_CSV=ON -DARROW_HDFS=ON -DARROW_FILESYSTEM=ON -DARROW_WITH_SNAPPY=ON -DARROW_JSON=ON -DARROW_DATASET=ON -DARROW_WITH_LZ4=ON -DGTEST_ROOT=/usr/src/gtest && make -j2
           sudo make install
@@ -88,7 +88,7 @@ jobs:
         run: |
           cd /tmp
           git clone https://github.com/oap-project/arrow.git
-          cd arrow && git checkout arrow-4.0.0-oap && cd cpp
+          cd arrow && git checkout arrow-7.0.0-oap && cd cpp
           mkdir build && cd build
           cmake .. -DARROW_JNI=ON -DARROW_GANDIVA_JAVA=ON -DARROW_GANDIVA=ON -DARROW_PARQUET=ON -DARROW_ORC=ON -DARROW_CSV=ON -DARROW_HDFS=ON -DARROW_FILESYSTEM=ON -DARROW_WITH_SNAPPY=ON -DARROW_JSON=ON -DARROW_DATASET=ON -DARROW_WITH_LZ4=ON -DGTEST_ROOT=/usr/src/gtest && make -j2
           sudo make install
@@ -133,7 +133,7 @@ jobs:
         run: |
           cd /tmp
           git clone https://github.com/oap-project/arrow.git
-          cd arrow && git checkout arrow-4.0.0-oap && cd cpp
+          cd arrow && git checkout arrow-7.0.0-oap && cd cpp
           mkdir build && cd build
           cmake .. -DARROW_JNI=ON -DARROW_GANDIVA_JAVA=ON -DARROW_GANDIVA=ON -DARROW_PARQUET=ON -DARROW_ORC=ON -DARROW_CSV=ON -DARROW_HDFS=ON -DARROW_FILESYSTEM=ON -DARROW_WITH_SNAPPY=ON -DARROW_JSON=ON -DARROW_DATASET=ON -DARROW_WITH_LZ4=ON -DGTEST_ROOT=/usr/src/gtest && make -j2
           sudo make install

diff --git a/arrow-data-source/README.md b/arrow-data-source/README.md
@@ -117,7 +117,7 @@ You have to use a customized Arrow to support for our datasets Java API.
 
 ```
 // build arrow-cpp
-git clone -b arrow-4.0.0-oap https://github.com/oap-project/arrow.git
+git clone -b arrow-7.0.0-oap https://github.com/oap-project/arrow.git
 cd arrow/cpp
 mkdir build
 cd build

diff --git a/arrow-data-source/common/src/main/scala/com/intel/oap/spark/sql/ArrowWriteQueue.scala b/arrow-data-source/common/src/main/scala/com/intel/oap/spark/sql/ArrowWriteQueue.scala
@@ -27,7 +27,6 @@ import java.util.regex.Pattern
 
 import com.intel.oap.spark.sql.ArrowWriteQueue.EOS_BATCH
 import com.intel.oap.spark.sql.ArrowWriteQueue.ScannerImpl
-import org.apache.arrow.dataset.file.DatasetFileWriter
 import org.apache.arrow.dataset.file.format.FileFormat
 import org.apache.arrow.dataset.scanner.Scanner
 import org.apache.arrow.dataset.scanner.ScanTask
@@ -47,12 +46,15 @@ class ArrowWriteQueue(schema: Schema, fileFormat: FileFormat, outputFileURI: Str
     val dirURI = matcher.group(1)
     val fileName = matcher.group(2)
 
-    DatasetFileWriter.write(scanner, fileFormat, dirURI, Array(), 1, fileName)
+//    disable write by arrow 7.0.0
+//    DatasetFileWriter.write(scanner, fileFormat, dirURI, Array(), 1, fileName)
   }, "ArrowWriteQueue - " + UUID.randomUUID().toString)
 
   writeThread.start()
 
   def enqueue(batch: ArrowRecordBatch): Unit = {
+    //    disable write by arrow 7.0.0
+    throw new UnsupportedOperationException("write is disabled by arrow 7.0.0 rebase")
     scanner.enqueue(batch)
   }
 

diff --git a/arrow-data-source/script/build_arrow.sh b/arrow-data-source/script/build_arrow.sh
@@ -62,7 +62,7 @@ echo "ARROW_SOURCE_DIR=${ARROW_SOURCE_DIR}"
 echo "ARROW_INSTALL_DIR=${ARROW_INSTALL_DIR}"
 mkdir -p $ARROW_SOURCE_DIR
 mkdir -p $ARROW_INSTALL_DIR
-git clone https://github.com/oap-project/arrow.git  --branch arrow-4.0.0-oap $ARROW_SOURCE_DIR
+git clone https://github.com/oap-project/arrow.git  --branch arrow-7.0.0-oap $ARROW_SOURCE_DIR
 pushd $ARROW_SOURCE_DIR
 
 cmake ./cpp \
@@ -98,7 +98,7 @@ make -j$NPROC
 make install
 
 cd java        
-mvn clean install -P arrow-jni -am -Darrow.cpp.build.dir=${ARROW_INSTALL_DIR}/lib -DskipTests -Dcheckstyle.skip
+mvn clean install -P arrow-jni -pl dataset,gandiva -am -Darrow.cpp.build.dir=${ARROW_INSTALL_DIR}/lib -DskipTests -Dcheckstyle.skip
 popd
 echo "Finish to build Arrow from Source !!!"
 else

diff --git a/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowFileFormat.scala b/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowFileFormat.scala
@@ -23,7 +23,7 @@ import scala.collection.JavaConverters._
 
 import com.intel.oap.spark.sql.ArrowWriteExtension.FakeRow
 import com.intel.oap.spark.sql.ArrowWriteQueue
-import com.intel.oap.spark.sql.execution.datasources.v2.arrow.{ArrowFilters, ArrowOptions, ArrowUtils}
+import com.intel.oap.spark.sql.execution.datasources.v2.arrow.{ArrowOptions, ArrowUtils}
 import com.intel.oap.spark.sql.execution.datasources.v2.arrow.ArrowSQLConf._
 import com.intel.oap.vectorized.ArrowWritableColumnVector
 import org.apache.arrow.dataset.scanner.ScanOptions
@@ -128,7 +128,8 @@ class ArrowFileFormat extends FileFormat with DataSourceRegister with Serializab
       val dataset = factory.finish();
 
       val filter = if (enableFilterPushDown) {
-        ArrowFilters.translateFilters(filters)
+        // disable filter pushdown by arrow 7.0.0
+        org.apache.arrow.dataset.filter.Filter.EMPTY
       } else {
         org.apache.arrow.dataset.filter.Filter.EMPTY
       }

diff --git a/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowFilters.scala b/.../src/main/scala/com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowFilters.scala
diff --git a/.../com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowPartitionReaderFactory.scala b/.../com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowPartitionReaderFactory.scala
@@ -61,7 +61,8 @@ case class ArrowPartitionReaderFactory(
       partitionedFile.start, partitionedFile.length, options)
     val dataset = factory.finish()
     val filter = if (enableFilterPushDown) {
-      ArrowFilters.translateFilters(ArrowFilters.pruneWithSchema(pushedFilters, readDataSchema))
+      // disable filter pushdown by arrow 7.0.0
+      org.apache.arrow.dataset.filter.Filter.EMPTY
     } else {
       org.apache.arrow.dataset.filter.Filter.EMPTY
     }

diff --git a/...rd/src/main/scala/com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowUtils.scala b/...rd/src/main/scala/com/intel/oap/spark/sql/execution/datasources/v2/arrow/ArrowUtils.scala
@@ -117,7 +117,8 @@ object ArrowUtils {
     val paramMap = options.parameters.toMap.asJava
     options.originalFormat match {
       case "parquet" => org.apache.arrow.dataset.file.format.ParquetFileFormat.create(paramMap)
-      case "orc" => org.apache.arrow.dataset.file.format.OrcFileFormat.create(paramMap)
+//      disable orc by arrow 7.0.0
+//      case "orc" => org.apache.arrow.dataset.file.format.OrcFileFormat.create(paramMap)
       case "csv" => org.apache.arrow.dataset.file.format.CsvFileFormat.create(paramMap)
       case _ => throw new IllegalArgumentException("Unrecognizable format")
     }

diff --git a/.../test/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowDataSourceTest.scala b/.../test/scala/com/intel/oap/spark/sql/execution/datasources/arrow/ArrowDataSourceTest.scala
@@ -203,7 +203,7 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
     assert(rows.length === 3)
   }
 
-  test("simple parquet write") {
+  ignore("simple parquet write") {
     val path = ArrowDataSourceTest.locateResourcePath(parquetFile3)
     val frame = spark.read
         .option(ArrowOptions.KEY_ORIGINAL_FORMAT, "parquet")
@@ -339,7 +339,7 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
     df.show()
   }
 
-  test("orc reader on data type: struct, array, map") {
+  ignore("orc reader on data type: struct, array, map") {
     val path = ArrowDataSourceTest.locateResourcePath(orcFile1)
     val frame = spark.read
         .option(ArrowOptions.KEY_ORIGINAL_FORMAT, "orc")
@@ -363,7 +363,7 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
   }
 
   private val orcFile = "people.orc"
-  test("read orc file") {
+  ignore("read orc file") {
     val path = ArrowDataSourceTest.locateResourcePath(orcFile)
     verifyFrame(
       spark.read
@@ -372,15 +372,15 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
           .load(path), 2, 3)
   }
 
-  test("read orc file - programmatic API ") {
+  ignore("read orc file - programmatic API ") {
     val path = ArrowDataSourceTest.locateResourcePath(orcFile)
     verifyFrame(
       spark.read
           .option(ArrowOptions.KEY_ORIGINAL_FORMAT, "orc")
           .arrow(path), 2, 3)
   }
 
-  test("create catalog table for orc") {
+  ignore("create catalog table for orc") {
     val path = ArrowDataSourceTest.locateResourcePath(orcFile)
     //    spark.catalog.createTable("people", path, "arrow")
     spark.catalog.createTable("people", "arrow", Map("path" -> path, "originalFormat" -> "orc"))
@@ -389,7 +389,7 @@ class ArrowDataSourceTest extends QueryTest with SharedSparkSession {
     verifyFrame(spark.sql(sql), 2, 3)
   }
 
-  test("simple SQL query on orc file ") {
+  ignore("simple SQL query on orc file ") {
     val path = ArrowDataSourceTest.locateResourcePath(orcFile)
     val frame = spark.read
         .option(ArrowOptions.KEY_ORIGINAL_FORMAT, "orc")

diff --git a/docs/ApacheArrowInstallation.md b/docs/ApacheArrowInstallation.md
@@ -30,7 +30,7 @@ Please make sure your cmake version is qualified based on the prerequisite.
 # Arrow
 ``` shell
 git clone https://github.com/oap-project/arrow.git
-cd arrow && git checkout arrow-4.0.0-oap
+cd arrow && git checkout arrow-7.0.0-oap
 mkdir -p arrow/cpp/release-build
 cd arrow/cpp/release-build
 cmake -DARROW_DEPENDENCY_SOURCE=BUNDLED -DARROW_GANDIVA_JAVA=ON -DARROW_GANDIVA=ON -DARROW_PARQUET=ON -DARROW_ORC=ON -DARROW_CSV=ON -DARROW_HDFS=ON -DARROW_BOOST_USE_SHARED=ON -DARROW_JNI=ON -DARROW_DATASET=ON -DARROW_WITH_PROTOBUF=ON -DARROW_WITH_SNAPPY=ON -DARROW_WITH_LZ4=ON -DARROW_FILESYSTEM=ON -DARROW_JSON=ON ..