diff --git a/jvm/src/main/java/com/intel/oap/row/RowIterator.java b/jvm/src/main/java/com/intel/oap/row/RowIterator.java index ced689741b3..86fb78b8e3b 100644 --- a/jvm/src/main/java/com/intel/oap/row/RowIterator.java +++ b/jvm/src/main/java/com/intel/oap/row/RowIterator.java @@ -17,41 +17,45 @@ package com.intel.oap.row; -import com.intel.oap.vectorized.JniInstance; -import io.kyligence.jni.engine.LocalEngine; +import com.intel.oap.vectorized.MetricsObject; import java.io.IOException; public class RowIterator { + private native boolean nativeHasNext(long nativeHandler); + private native SparkRowInfo nativeNext(long nativeHandler); + private native void nativeClose(long nativeHandler); + private native MetricsObject nativeFetchMetrics(long nativeHandler); - private LocalEngine localEngine; + private long nativeHandler = 0L; private boolean closed = false; - public RowIterator() throws IOException {} - - public RowIterator(byte[] plan, String soFilePath) throws IOException { - this.localEngine = JniInstance.getInstanceWithLibPath(soFilePath).buildLocalEngine(plan); - this.localEngine.execute(); + public RowIterator(long nativeHandler) throws IOException { + this.nativeHandler = nativeHandler; } public boolean hasNext() throws IOException { - return this.localEngine.hasNext(); + return nativeHasNext(nativeHandler); } public SparkRowInfo next() throws IOException { - if (this.localEngine == null) { + if (nativeHandler == 0) { + return null; + } + return nativeNext(nativeHandler); + } + + public MetricsObject getMetrics() throws IOException, ClassNotFoundException { + if (nativeHandler == 0) { return null; } - return this.localEngine.next(); + return nativeFetchMetrics(nativeHandler); } public void close() { if (!closed) { - if (this.localEngine != null) { - try { - this.localEngine.close(); - } catch (IOException e) { - } + if (nativeHandler != 0L) { + nativeClose(nativeHandler); } closed = true; } diff --git a/jvm/src/main/java/com/intel/oap/vectorized/BatchIterator.java b/jvm/src/main/java/com/intel/oap/vectorized/BatchIterator.java index 3a7327a8658..895773f53ed 100644 --- a/jvm/src/main/java/com/intel/oap/vectorized/BatchIterator.java +++ b/jvm/src/main/java/com/intel/oap/vectorized/BatchIterator.java @@ -43,7 +43,6 @@ public class BatchIterator implements AutoCloseable, Serializable { public BatchIterator() throws IOException {} public BatchIterator(long instance_id) throws IOException { - JniInstance.getInstance(); nativeHandler = instance_id; } diff --git a/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java b/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java index 32e2646df31..9d5cfc1a08f 100644 --- a/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java +++ b/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java @@ -19,6 +19,7 @@ import com.intel.oap.GazelleJniConfig; import com.intel.oap.execution.ColumnarNativeIterator; +import com.intel.oap.row.RowIterator; import com.intel.oap.substrait.plan.PlanNode; import org.apache.arrow.dataset.jni.NativeMemoryPool; import org.apache.arrow.gandiva.exceptions.GandivaException; @@ -46,20 +47,29 @@ public ExpressionEvaluator() throws IOException, IllegalAccessException, Illegal } public ExpressionEvaluator(List listJars) throws IOException, IllegalAccessException, IllegalStateException { - this(listJars, null); + this(listJars, GazelleJniConfig.getConf().nativeLibName()); } public ExpressionEvaluator(List listJars, String libName) throws IOException, IllegalAccessException, IllegalStateException { - this(listJars, libName, true); + this(listJars, libName, + GazelleJniConfig.getSessionConf().nativeLibPath(), + GazelleJniConfig.getConf().loadArrow()); } - public ExpressionEvaluator(List listJars, String libName, boolean loadArrowAndGandiva) + public ExpressionEvaluator(String libPath) + throws IOException, IllegalAccessException, IllegalStateException { + this(java.util.Collections.emptyList(), null, libPath, GazelleJniConfig.getConf().loadArrow()); + } + + public ExpressionEvaluator(List listJars, String libName, + String libPath, boolean loadArrowAndGandiva) throws IOException, IllegalAccessException, IllegalStateException { String tmp_dir = GazelleJniConfig.getTempFile(); if (tmp_dir == null) { tmp_dir = System.getProperty("java.io.tmpdir"); } - jniWrapper = new ExpressionEvaluatorJniWrapper(tmp_dir, listJars, libName, loadArrowAndGandiva); + jniWrapper = new ExpressionEvaluatorJniWrapper(tmp_dir, listJars, libName, libPath, + loadArrowAndGandiva); jniWrapper.nativeSetJavaTmpDir(jniWrapper.tmp_dir_path); jniWrapper.nativeSetBatchSize(GazelleJniConfig.getBatchSize()); jniWrapper.nativeSetMetricsTime(GazelleJniConfig.getEnableMetricsTime()); @@ -70,8 +80,12 @@ long getInstanceId() { return nativeHandler; } + public void initNative() { + jniWrapper.nativeInitNative(); + } + /** Used by WholeStageTransfrom */ - public BatchIterator createKernelWithIterator( + public BatchIterator createKernelWithBatchIterator( byte[] wsPlan, ArrayList iterList) throws RuntimeException, IOException { NativeMemoryPool memoryPool = SparkMemoryUtils.contextMemoryPool(); @@ -81,10 +95,17 @@ public BatchIterator createKernelWithIterator( return new BatchIterator(batchIteratorInstance); } - public BatchIterator createKernelWithIterator( + public BatchIterator createKernelWithBatchIterator( PlanNode wsPlan, ArrayList iterList) throws RuntimeException, IOException { - return createKernelWithIterator(getPlanBytesBuf(wsPlan), iterList); + return createKernelWithBatchIterator(getPlanBytesBuf(wsPlan), iterList); + } + + public RowIterator createKernelWithRowIterator( + byte[] wsPlan, + ArrayList iterList) throws RuntimeException, IOException { + long rowIteratorInstance = jniWrapper.nativeCreateKernelWithRowIterator(wsPlan); + return new RowIterator(rowIteratorInstance); } @Override diff --git a/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluatorJniWrapper.java b/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluatorJniWrapper.java index 095394e0bdc..efc95629073 100644 --- a/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluatorJniWrapper.java +++ b/jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluatorJniWrapper.java @@ -34,23 +34,34 @@ public class ExpressionEvaluatorJniWrapper { /** Wrapper for native API. */ public ExpressionEvaluatorJniWrapper(String tmp_dir, List listJars, String libName, - boolean loadArrowAndGandiva) + String libPath, boolean loadArrowAndGandiva) throws IOException, IllegalAccessException, IllegalStateException { - JniInstance jni = JniInstance.getInstance(tmp_dir, libName, loadArrowAndGandiva); + JniInstance jni = JniInstance.getInstance(tmp_dir, libName, libPath, + loadArrowAndGandiva); jni.setTempDir(); jni.setJars(listJars); tmp_dir_path = jni.getTempDir(); } /** - * Create a whole_stage_transfrom kernel, and return a result iterator. + * Call initNative to initialize native computing. + */ + native void nativeInitNative(); + + /** + * Create a whole_stage_transform kernel, and return a result iterator. * * @param nativeHandler nativeHandler of this expression * @return iterator instance id */ native long nativeCreateKernelWithIterator(long nativeHandler, byte[] wsExprListBuf, - ColumnarNativeIterator[] batchItr)throws RuntimeException; + ColumnarNativeIterator[] batchItr) throws RuntimeException; + + /** + * Create a whole stage transform kernel and return a row iterator. + */ + native long nativeCreateKernelWithRowIterator(byte[] wsPlan) throws RuntimeException; /** * Set native env variables NATIVE_TMP_DIR @@ -69,9 +80,6 @@ native long nativeCreateKernelWithIterator(long nativeHandler, /** * Set native env variables NATIVESQL_METRICS_TIME - * - * @param batch_size numRows of one batch, use - * spark.sql.execution.arrow.maxRecordsPerBatch */ native void nativeSetMetricsTime(boolean is_enable); diff --git a/jvm/src/main/java/com/intel/oap/vectorized/JniInstance.java b/jvm/src/main/java/com/intel/oap/vectorized/JniInstance.java index 748412eccac..b0caa50c91b 100644 --- a/jvm/src/main/java/com/intel/oap/vectorized/JniInstance.java +++ b/jvm/src/main/java/com/intel/oap/vectorized/JniInstance.java @@ -17,7 +17,6 @@ package com.intel.oap.vectorized; -import io.kyligence.jni.engine.LocalEngine; import org.apache.commons.lang3.StringUtils; import java.io.*; @@ -58,16 +57,16 @@ public static JniInstance getInstance() throws IOException { } public static JniInstance getInstance(String tmp_dir) throws IOException { - return getInstance(tmp_dir, null, true); + return getInstance(tmp_dir, null, null, true); } - public static JniInstance getInstance(String tmp_dir, String lib_name, + public static JniInstance getInstance(String tmp_dir, String lib_name, String libPath, boolean loadArrowAndGandiva) throws IOException { if (INSTANCE == null) { synchronized (JniInstance.class) { if (INSTANCE == null) { try { - INSTANCE = new JniInstance(tmp_dir, lib_name, loadArrowAndGandiva); + INSTANCE = new JniInstance(tmp_dir, lib_name, libPath, loadArrowAndGandiva); } catch (IllegalAccessException ex) { throw new IOException("IllegalAccess", ex); } @@ -77,61 +76,8 @@ public static JniInstance getInstance(String tmp_dir, String lib_name, return INSTANCE; } - public static JniInstance getInstanceWithLibPath() { - return getInstanceWithLibPath(null); - } - - /** - * A function used to load an pre-installed library from the specific path. - * **/ - public static synchronized JniInstance getInstanceWithLibPath(String libPath) { - if (INSTANCE != null) { - return INSTANCE; - } - File file = null; - boolean libPathExists = false; - if (StringUtils.isNotBlank(libPath)) { - file = new File(libPath); - libPathExists = file.isFile() && file.exists(); - } - String libName = "liblocal_engine_jnid"; - if (!libPathExists) { - String soFileName = "/" + libName + ".so"; - try { - InputStream is = JniInstance.class.getResourceAsStream(soFileName); - file = File.createTempFile("lib", ".so"); - OutputStream os = new FileOutputStream(file); - byte[] buffer = new byte[128 << 10]; - int length; - while ((length = is.read(buffer)) != -1) { - os.write(buffer, 0, length); - } - is.close(); - os.close(); - } catch (IOException e) { - } - } - if (file != null) { - try { - file.setReadable(true, false); - System.load(file.getAbsolutePath()); - libPath = file.getAbsolutePath(); - } catch (UnsatisfiedLinkError error) { - throw error; - } - } - INSTANCE = new JniInstance(); - LocalEngine.initEngineEnv(); - return INSTANCE; - } - - public LocalEngine buildLocalEngine(byte[] substraitPlan) { - return new LocalEngine(substraitPlan); - } - - private JniInstance() throws IllegalStateException {} - - private JniInstance(String _tmp_dir, String _lib_name, boolean loadArrowAndGandiva) + private JniInstance(String _tmp_dir, String _lib_name, String libPath, + boolean loadArrowAndGandiva) throws IOException, IllegalAccessException, IllegalStateException { if (!isLoaded) { if (_tmp_dir.contains("nativesql")) { @@ -142,7 +88,11 @@ private JniInstance(String _tmp_dir, String _lib_name, boolean loadArrowAndGandi tmp_dir = path.toAbsolutePath().toString(); } try { - loadLibraryFromJarWithLib(tmp_dir, _lib_name, loadArrowAndGandiva); + if (StringUtils.isNotBlank(libPath)) { + loadLibraryFromLibPath(tmp_dir, libPath, loadArrowAndGandiva); + } else { + loadLibraryFromJarWithLib(tmp_dir, _lib_name, loadArrowAndGandiva); + } } catch (IOException ex) { if (_lib_name != null) { System.loadLibrary(_lib_name); @@ -209,6 +159,48 @@ private static void loadLibraryFromJar(String source_jar, String tmp_dir) throws } } + /** + * A function used to load arrow and gandiva lib from jars + */ + static void loadArrowAndGandivaFromJarWithLib(String tmp_dir) throws IOException, IllegalAccessException { + final File arrowlibraryFile = moveFileFromJarToTemp(tmp_dir, ARROW_LIBRARY_NAME); + Path arrow_target = Paths.get(arrowlibraryFile.getPath()); + Path arrow_link = Paths.get(tmp_dir, ARROW_PARENT_LIBRARY_NAME); + if (Files.exists(arrow_link)) { + Files.delete(arrow_link); + } + Path symLink = Files.createSymbolicLink(arrow_link, arrow_target); + System.load(arrowlibraryFile.getAbsolutePath()); + + final File gandivalibraryFile = moveFileFromJarToTemp(tmp_dir, GANDIVA_LIBRARY_NAME); + Path gandiva_target = Paths.get(gandivalibraryFile.getPath()); + Path gandiva_link = Paths.get(tmp_dir, GANDIVA_PARENT_LIBRARY_NAME); + if (Files.exists(gandiva_link)) { + Files.delete(gandiva_link); + } + Files.createSymbolicLink(gandiva_link, gandiva_target); + System.load(gandivalibraryFile.getAbsolutePath()); + } + + /** + * A function used to load an pre-installed library from the specific path. + * **/ + static void loadLibraryFromLibPath(String tmp_dir, String libPath, + boolean loadArrowAndGandiva) + throws IOException, IllegalAccessException { + synchronized (JniInstance.class) { + File file = new File(libPath); + if (!file.isFile() || !file.exists()) { + throw new IOException("Library path: " + libPath + " is not a file or does not exist."); + } + System.load(file.getAbsolutePath()); + + if (loadArrowAndGandiva) { + loadArrowAndGandivaFromJarWithLib(tmp_dir); + } + } + } + static void loadLibraryFromJarWithLib(String tmp_dir, String lib_name, boolean loadArrowAndGandiva) throws IOException, IllegalAccessException { synchronized (JniInstance.class) { @@ -226,23 +218,7 @@ static void loadLibraryFromJarWithLib(String tmp_dir, String lib_name, System.load(libraryFile.getAbsolutePath()); if (loadArrowAndGandiva) { - final File arrowlibraryFile = moveFileFromJarToTemp(tmp_dir, ARROW_LIBRARY_NAME); - Path arrow_target = Paths.get(arrowlibraryFile.getPath()); - Path arrow_link = Paths.get(tmp_dir, ARROW_PARENT_LIBRARY_NAME); - if (Files.exists(arrow_link)) { - Files.delete(arrow_link); - } - Path symLink = Files.createSymbolicLink(arrow_link, arrow_target); - System.load(arrowlibraryFile.getAbsolutePath()); - - final File gandivalibraryFile = moveFileFromJarToTemp(tmp_dir, GANDIVA_LIBRARY_NAME); - Path gandiva_target = Paths.get(gandivalibraryFile.getPath()); - Path gandiva_link = Paths.get(tmp_dir, GANDIVA_PARENT_LIBRARY_NAME); - if (Files.exists(gandiva_link)) { - Files.delete(gandiva_link); - } - Files.createSymbolicLink(gandiva_link, gandiva_target); - System.load(gandivalibraryFile.getAbsolutePath()); + loadArrowAndGandivaFromJarWithLib(tmp_dir); } } } diff --git a/jvm/src/main/java/io/kyligence/jni/engine/LocalEngine.java b/jvm/src/main/java/io/kyligence/jni/engine/LocalEngine.java deleted file mode 100644 index d3c7fd87dd2..00000000000 --- a/jvm/src/main/java/io/kyligence/jni/engine/LocalEngine.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.kyligence.jni.engine; - -import java.io.Closeable; -import java.io.IOException; - -import com.intel.oap.row.SparkRowInfo; - -public class LocalEngine implements Closeable { - public static native long test(int a, int b); - - public static native void initEngineEnv(); - - private long nativeExecutor; - private byte[] plan; - - public LocalEngine(byte[] plan) { - this.plan = plan; - } - - public native void execute(); - - public native boolean hasNext(); - - public native SparkRowInfo next(); - - - @Override - public native void close() throws IOException; -} diff --git a/jvm/src/main/scala/com/intel/oap/GazelleJniConfig.scala b/jvm/src/main/scala/com/intel/oap/GazelleJniConfig.scala index 6c868d31899..65aa5685a24 100644 --- a/jvm/src/main/scala/com/intel/oap/GazelleJniConfig.scala +++ b/jvm/src/main/scala/com/intel/oap/GazelleJniConfig.scala @@ -17,6 +17,8 @@ package com.intel.oap +import java.util.Locale + import org.apache.spark.internal.Logging import org.apache.spark.sql.internal.SQLConf @@ -27,15 +29,20 @@ case class GazelleNumaBindingInfo( class GazelleJniConfig(conf: SQLConf) extends Logging { def getCpu: Boolean = { - val source = scala.io.Source.fromFile("/proc/cpuinfo") - val lines = try source.mkString finally source.close() - // TODO(): check CPU flags to enable/disable AVX512 - if (lines.contains("GenuineIntel")) { + // only for developing on mac + if (System.getProperty("os.name").toLowerCase(Locale.ROOT).contains("mac")) { true } else { - // System.out.println(actualSchemaRoot.getRowCount()); - logWarning("running on non-intel CPU, disable all columnar operators") - false + val source = scala.io.Source.fromFile("/proc/cpuinfo") + val lines = try source.mkString finally source.close() + // TODO(): check CPU flags to enable/disable AVX512 + if (lines.contains("GenuineIntel")) { + true + } else { + // System.out.println(actualSchemaRoot.getRowCount()); + logWarning("running on non-intel CPU, disable all columnar operators") + false + } } } @@ -133,21 +140,21 @@ class GazelleJniConfig(conf: SQLConf) extends Logging { // This config is used for deciding whether to load the native library. // When false, only Java code will be executed for a quick test. val loadNative: Boolean = - conf.getConfString("spark.oap.sql.columnar.loadnative", "true").toBoolean + conf.getConfString(GazelleJniConfig.OAP_LOAD_NATIVE, "true").toBoolean // This config is used for deciding whether to load Arrow and Gandiva libraries from // the native library. If the native library does not depend on Arrow and Gandiva, // this config should will set as false. val loadArrow: Boolean = - conf.getConfString("spark.oap.sql.columnar.loadarrow", "true").toBoolean + conf.getConfString(GazelleJniConfig.OAP_LOAD_ARROW, "true").toBoolean // This config is used for specifying the name of the native library. val nativeLibName: String = - conf.getConfString("spark.oap.sql.columnar.libname", "spark_columnar_jni") + conf.getConfString(GazelleJniConfig.OAP_LIB_NAME, "spark_columnar_jni") // This config is used for specifying the absolute path of the native library. val nativeLibPath: String = - conf.getConfString("spark.oap.sql.columnar.libpath", "") + conf.getConfString(GazelleJniConfig.OAP_LIB_PATH, "") // fallback to row operators if there are several continous joins val joinOptimizationThrottle: Integer = @@ -199,6 +206,12 @@ class GazelleJniConfig(conf: SQLConf) extends Logging { } object GazelleJniConfig { + + val OAP_LOAD_NATIVE = "spark.oap.sql.columnar.loadnative" + val OAP_LIB_NAME = "spark.oap.sql.columnar.libname" + val OAP_LIB_PATH = "spark.oap.sql.columnar.libpath" + val OAP_LOAD_ARROW = "spark.oap.sql.columnar.loadarrow" + var ins: GazelleJniConfig = null var random_temp_dir_path: String = null diff --git a/jvm/src/main/scala/com/intel/oap/GazellePlugin.scala b/jvm/src/main/scala/com/intel/oap/GazellePlugin.scala index be46d36bd11..09aaa772b96 100644 --- a/jvm/src/main/scala/com/intel/oap/GazellePlugin.scala +++ b/jvm/src/main/scala/com/intel/oap/GazellePlugin.scala @@ -17,20 +17,17 @@ package com.intel.oap -import java.util -import java.util.Collections -import java.util.Objects +import java.util.{Collections, Objects} import scala.language.implicitConversions -import com.intel.oap.GazellePlugin.GAZELLE_SESSION_EXTENSION_NAME -import com.intel.oap.GazellePlugin.SPARK_SESSION_EXTS_KEY + +import com.intel.oap.GazellePlugin.{GAZELLE_SESSION_EXTENSION_NAME, SPARK_SESSION_EXTS_KEY} import com.intel.oap.extension.{ColumnarOverrides, StrategyOverrides} -import org.apache.spark.SparkConf -import org.apache.spark.SparkContext -import org.apache.spark.api.plugin.DriverPlugin -import org.apache.spark.api.plugin.ExecutorPlugin -import org.apache.spark.api.plugin.PluginContext -import org.apache.spark.api.plugin.SparkPlugin +import com.intel.oap.vectorized.ExpressionEvaluator +import java.util +import org.apache.spark.{SparkConf, SparkContext} + +import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin, PluginContext, SparkPlugin} import org.apache.spark.sql.SparkSessionExtensions import org.apache.spark.sql.internal.StaticSQLConf @@ -61,7 +58,27 @@ private[oap] class GazelleDriverPlugin extends DriverPlugin { } private[oap] class GazelleExecutorPlugin extends ExecutorPlugin { - // N/A + /** + * Initialize the executor plugin. + */ + override def init(ctx: PluginContext, extraConf: util.Map[String, String]): Unit = { + // SQLConf is not initialed here, so it can not use 'GazelleJniConfig.getConf' to get conf. + if (ctx.conf().getBoolean(GazelleJniConfig.OAP_LOAD_NATIVE, true)) { + val initKernel = new ExpressionEvaluator(java.util.Collections.emptyList[String], + ctx.conf().get(GazelleJniConfig.OAP_LIB_NAME, "spark_columnar_jni"), + ctx.conf().get(GazelleJniConfig.OAP_LIB_PATH, ""), + ctx.conf().getBoolean(GazelleJniConfig.OAP_LOAD_ARROW, true)) + initKernel.initNative() + } + } + + /** + * Clean up and terminate this plugin. + * For example: close the native engine. + */ + override def shutdown(): Unit = { + super.shutdown() + } } private[oap] class GazelleSessionExtensions extends (SparkSessionExtensions => Unit) { diff --git a/jvm/src/main/scala/com/intel/oap/execution/NativeWholeStageColumnarRDD.scala b/jvm/src/main/scala/com/intel/oap/execution/NativeWholeStageColumnarRDD.scala index 6bb087ed26d..e672f807997 100644 --- a/jvm/src/main/scala/com/intel/oap/execution/NativeWholeStageColumnarRDD.scala +++ b/jvm/src/main/scala/com/intel/oap/execution/NativeWholeStageColumnarRDD.scala @@ -18,24 +18,24 @@ package com.intel.oap.execution import java.io.Serializable -import java.util import scala.collection.JavaConverters._ import scala.collection.mutable import scala.collection.mutable.ListBuffer -import com.google.common.collect.Lists + import com.intel.oap.GazelleJniConfig import com.intel.oap.expression.ConverterUtils import com.intel.oap.vectorized._ import org.apache.arrow.vector.types.pojo.Schema import org.apache.spark._ + import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory} import org.apache.spark.sql.execution.datasources.PartitionedFile import org.apache.spark.sql.util.ArrowUtils import org.apache.spark.sql.util.OASPackageBridge._ -import org.apache.spark.sql.vectorized.{ColumnVector, ColumnarBatch} +import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector} import org.apache.spark.util._ case class NativeFilePartition(index: Int, files: Array[PartitionedFile], @@ -74,8 +74,6 @@ class NativeWholeStageColumnarRDD( extends RDD[ColumnarBatch](sc, Nil) { val numaBindingInfo = GazelleJniConfig.getConf.numaBindingInfo val loadNative: Boolean = GazelleJniConfig.getConf.loadNative - val libName: String = GazelleJniConfig.getConf.nativeLibName - val loadArrow: Boolean = GazelleJniConfig.getConf.loadArrow override protected def getPartitions: Array[Partition] = { inputPartitions.zipWithIndex.map { @@ -103,10 +101,10 @@ class NativeWholeStageColumnarRDD( var resIter : BatchIterator = null if (loadNative) { // TODO: 'jarList' is kept for codegen - val transKernel = new ExpressionEvaluator(jarList.toList.asJava, libName, loadArrow) + val transKernel = new ExpressionEvaluator(jarList.toList.asJava) val inBatchIters = new java.util.ArrayList[ColumnarNativeIterator]() outputSchema = ConverterUtils.toArrowSchema(outputAttributes) - resIter = transKernel.createKernelWithIterator( + resIter = transKernel.createKernelWithBatchIterator( inputPartition.substraitPlan, inBatchIters) } val iter = new Iterator[Any] { diff --git a/jvm/src/main/scala/com/intel/oap/execution/NativeWholestageRowRDD.scala b/jvm/src/main/scala/com/intel/oap/execution/NativeWholestageRowRDD.scala index 3544754c072..e4a73e07028 100644 --- a/jvm/src/main/scala/com/intel/oap/execution/NativeWholestageRowRDD.scala +++ b/jvm/src/main/scala/com/intel/oap/execution/NativeWholestageRowRDD.scala @@ -21,6 +21,7 @@ import java.io._ import com.intel.oap.GazelleJniConfig import com.intel.oap.row.RowIterator +import com.intel.oap.vectorized.ExpressionEvaluator import org.apache.spark.{Partition, SparkContext, SparkException, TaskContext} import org.apache.spark.rdd.RDD @@ -37,7 +38,6 @@ class NativeWholestageRowRDD( extends RDD[InternalRow](sc, Nil) { val numaBindingInfo = GazelleJniConfig.getConf.numaBindingInfo val loadNative = GazelleJniConfig.getConf.loadNative - val libName = GazelleJniConfig.getConf.nativeLibName override protected def getPartitions: Array[Partition] = { inputPartitions.zipWithIndex.map { @@ -62,8 +62,9 @@ class NativeWholestageRowRDD( var resIter : RowIterator = null if (loadNative) { - resIter = new RowIterator(inputPartition.substraitPlan, - GazelleJniConfig.getConf.nativeLibPath) + val transKernel = new ExpressionEvaluator() + val inBatchIters = new java.util.ArrayList[ColumnarNativeIterator]() + resIter = transKernel.createKernelWithRowIterator(inputPartition.substraitPlan, inBatchIters) } val iter = new Iterator[InternalRow] with AutoCloseable { diff --git a/jvm/src/main/scala/com/intel/oap/execution/WholeStageTransformerExec.scala b/jvm/src/main/scala/com/intel/oap/execution/WholeStageTransformerExec.scala index 7a196a321bb..0811b5d69ee 100644 --- a/jvm/src/main/scala/com/intel/oap/execution/WholeStageTransformerExec.scala +++ b/jvm/src/main/scala/com/intel/oap/execution/WholeStageTransformerExec.scala @@ -87,8 +87,6 @@ case class WholeStageTransformerExec(child: SparkPlan)(val transformStageId: Int val numaBindingInfo = GazelleJniConfig.getConf.numaBindingInfo val enableColumnarSortMergeJoinLazyRead: Boolean = GazelleJniConfig.getConf.enableColumnarSortMergeJoinLazyRead - val libName: String = GazelleJniConfig.getConf.nativeLibName - val loadArrow: Boolean = GazelleJniConfig.getConf.loadArrow override lazy val metrics = Map( "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"), @@ -330,7 +328,7 @@ case class WholeStageTransformerExec(child: SparkPlan)(val transformStageId: Int s"Generated substrait plan tooks: ${(System.nanoTime() - startTime) / 1000000} ms") val wsRDD = new NativeWholestageRowRDD(sparkContext, substraitPlanPartition, - batchScan.readerFactory, true) + batchScan.readerFactory, false) wsRDD } else { sparkContext.emptyRDD @@ -431,7 +429,7 @@ case class WholeStageTransformerExec(child: SparkPlan)(val transformStageId: Int TreeBuilder.makeExpression( lazyReadFunction, Field.nullable("result", new ArrowType.Int(32, true))) - val transKernel = new ExpressionEvaluator(jarList.toList.asJava, libName, loadArrow) + val transKernel = new ExpressionEvaluator(jarList.toList.asJava) val inBatchIter = new ColumnarNativeIterator(iter.asJava) val inBatchIters = new java.util.ArrayList[ColumnarNativeIterator]() inBatchIters.add(inBatchIter); @@ -439,7 +437,7 @@ case class WholeStageTransformerExec(child: SparkPlan)(val transformStageId: Int val beforeBuild = System.nanoTime() val inputSchema = ConverterUtils.toArrowSchema(inputAttributes) val outputSchema = ConverterUtils.toArrowSchema(outputAttributes) - val nativeIterator = transKernel.createKernelWithIterator(rootNode, inBatchIters) + val nativeIterator = transKernel.createKernelWithBatchIterator(rootNode, inBatchIters) build_elapse += System.nanoTime() - beforeBuild val resultStructType = ArrowUtils.fromArrowSchema(outputSchema) val resIter = streamedSortPlan match { diff --git a/jvm/src/test/scala/com/intel/oap/benchmarks/BenchmarkTest.scala b/jvm/src/test/scala/com/intel/oap/benchmarks/BenchmarkTest.scala index 8f30278c4c6..6061ccfe9d1 100644 --- a/jvm/src/test/scala/com/intel/oap/benchmarks/BenchmarkTest.scala +++ b/jvm/src/test/scala/com/intel/oap/benchmarks/BenchmarkTest.scala @@ -22,6 +22,8 @@ import java.io.File import scala.collection.mutable.ArrayBuffer import scala.io.Source +import com.intel.oap.GazelleJniConfig + import org.apache.spark.sql.SparkSession object BenchmarkTest { @@ -36,7 +38,8 @@ object BenchmarkTest { val resourcePath = rootPath + "../../../src/test/resources/" val dataPath = resourcePath + "/tpch-data/" val queryPath = resourcePath + "/queries/" - (new File(dataPath).getAbsolutePath, "parquet", 10, false, queryPath + "q06.sql", "") + //(new File(dataPath).getAbsolutePath, "parquet", 1, false, queryPath + "q06.sql", "") + ("/data1/test_output/tpch-data-sf10", "parquet", 100, false, queryPath + "q06.sql", "") } val sqlStr = Source.fromFile(new File(sqlFilePath), "UTF-8") @@ -47,14 +50,14 @@ object BenchmarkTest { val sessionBuilder = if (!configed) { sessionBuilderTmp - .master("local[1]") + .master("local[3]") .config("spark.driver.memory", "4G") .config("spark.driver.memoryOverhead", "6G") .config("spark.serializer", "org.apache.spark.serializer.JavaSerializer") .config("spark.default.parallelism", 1) .config("spark.sql.shuffle.partitions", 1) .config("spark.sql.adaptive.enabled", "false") - .config("spark.sql.files.maxPartitionBytes", 128 << 10 << 10) // default is 128M + .config("spark.sql.files.maxPartitionBytes", 1024 << 10 << 10) // default is 128M .config("spark.sql.files.minPartitionNum", "1") .config("spark.sql.parquet.filterPushdown", "true") .config("spark.locality.wait", "0s") @@ -63,14 +66,15 @@ object BenchmarkTest { .config("spark.sql.sources.useV1SourceList", "avro") .config("spark.memory.fraction", "0.3") .config("spark.memory.storageFraction", "0.3") - .config("spark.sql.parquet.columnarReaderBatchSize", "20000") + //.config("spark.sql.parquet.columnarReaderBatchSize", "20000") .config("spark.plugins", "com.intel.oap.GazellePlugin") - .config("spark.sql.execution.arrow.maxRecordsPerBatch", "20000") + //.config("spark.sql.execution.arrow.maxRecordsPerBatch", "20000") .config("spark.oap.sql.columnar.columnartorow", "false") - .config("spark.oap.sql.columnar.loadnative", "false") - .config("spark.oap.sql.columnar.libpath", + .config(GazelleJniConfig.OAP_LOAD_NATIVE, "true") + .config(GazelleJniConfig.OAP_LOAD_ARROW, "false") + .config(GazelleJniConfig.OAP_LIB_PATH, "/home/myubuntu/Works/c_cpp_projects/Kyligence-ClickHouse/cmake-build-release/utils/local-engine/liblocal_engine_jni.so") - .config("spark.oap.sql.columnar.iterator", "true") + .config("spark.oap.sql.columnar.iterator", "false") //.config("spark.sql.planChangeLog.level", "info") .config("spark.sql.columnVector.offheap.enabled", "true") .config("spark.memory.offHeap.enabled", "true") @@ -81,7 +85,7 @@ object BenchmarkTest { val spark = sessionBuilder.getOrCreate() if (!configed) { - spark.sparkContext.setLogLevel("INFO") + spark.sparkContext.setLogLevel("WARN") } testSQL(spark, parquetFilesPath, fileFormat, executedCnt, sqlStr.mkString)