Skip to content

[GJ-49] Unify JNI interfaces #50

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
rui-mo merged 1 commit into from
Jan 26, 2022
Merged

[GJ-49] Unify JNI interfaces #50

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
36 changes: 20 additions & 16 deletions jvm/src/main/java/com/intel/oap/row/RowIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
1 change: 0 additions & 1 deletion jvm/src/main/java/com/intel/oap/vectorized/BatchIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand Down
35 changes: 28 additions & 7 deletions jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -46,20 +47,29 @@ public ExpressionEvaluator() throws IOException, IllegalAccessException, Illegal
}

public ExpressionEvaluator(List<String> listJars) throws IOException, IllegalAccessException, IllegalStateException {
this(listJars, null);
this(listJars, GazelleJniConfig.getConf().nativeLibName());
}

public ExpressionEvaluator(List<String> listJars, String libName) throws IOException, IllegalAccessException, IllegalStateException {
this(listJars, libName, true);
this(listJars, libName,
GazelleJniConfig.getSessionConf().nativeLibPath(),
GazelleJniConfig.getConf().loadArrow());
}

public ExpressionEvaluator(List<String> 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<String> 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());
Expand All @@ -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<ColumnarNativeIterator> iterList)
throws RuntimeException, IOException {
NativeMemoryPool memoryPool = SparkMemoryUtils.contextMemoryPool();
Expand All @@ -81,10 +95,17 @@ public BatchIterator createKernelWithIterator(
return new BatchIterator(batchIteratorInstance);
}

public BatchIterator createKernelWithIterator(
public BatchIterator createKernelWithBatchIterator(
PlanNode wsPlan, ArrayList<ColumnarNativeIterator> iterList)
throws RuntimeException, IOException {
return createKernelWithIterator(getPlanBytesBuf(wsPlan), iterList);
return createKernelWithBatchIterator(getPlanBytesBuf(wsPlan), iterList);
}

public RowIterator createKernelWithRowIterator(
byte[] wsPlan,
ArrayList<ColumnarNativeIterator> iterList) throws RuntimeException, IOException {
long rowIteratorInstance = jniWrapper.nativeCreateKernelWithRowIterator(wsPlan);
return new RowIterator(rowIteratorInstance);
}

@Override
Expand Down
22 changes: 15 additions & 7 deletions jvm/src/main/java/com/intel/oap/vectorized/ExpressionEvaluatorJniWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,23 +34,34 @@ public class ExpressionEvaluatorJniWrapper {

/** Wrapper for native API. */
public ExpressionEvaluatorJniWrapper(String tmp_dir, List<String> 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
Expand All @@ -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);

Expand Down
130 changes: 53 additions & 77 deletions jvm/src/main/java/com/intel/oap/vectorized/JniInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package com.intel.oap.vectorized;

import io.kyligence.jni.engine.LocalEngine;
import org.apache.commons.lang3.StringUtils;

import java.io.*;
Expand Down Expand Up @@ -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);
}
Expand All @@ -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")) {
Expand All @@ -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);
Expand Down Expand Up @@ -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) {
Expand All @@ -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);
}
}
}
Expand Down
Loading