You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2024/03/15 03:14:59 UTC
(doris) branch branch-2.0 updated: [fix](jni) remove 'push_down_predicates' and fix BE crash with decimal predicate (#32253) (#32255)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 50309c054d4 [fix](jni) remove 'push_down_predicates' and fix BE crash with decimal predicate (#32253) (#32255)
50309c054d4 is described below
commit 50309c054d463c096b5cc8d7c1aa5e69e30fef90
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Fri Mar 15 11:14:52 2024 +0800
[fix](jni) remove 'push_down_predicates' and fix BE crash with decimal predicate (#32253) (#32255)
bp #32253
---
be/src/vec/exec/jni_connector.cpp | 17 ++++++++++-------
be/src/vec/exec/jni_connector.h | 3 +++
.../main/java/org/apache/doris/avro/AvroJNIScanner.java | 3 +--
.../main/java/org/apache/doris/hudi/HudiJniScanner.java | 16 +---------------
.../java/org/apache/doris/common/jni/JniScanner.java | 12 +++++++++---
.../org/apache/doris/common/jni/MockJniScanner.java | 11 +----------
.../apache/doris/maxcompute/MaxComputeJniScanner.java | 16 +++-------------
.../java/org/apache/doris/paimon/PaimonJniScanner.java | 3 +--
8 files changed, 29 insertions(+), 52 deletions(-)
diff --git a/be/src/vec/exec/jni_connector.cpp b/be/src/vec/exec/jni_connector.cpp
index 73985b81c71..5c85e329022 100644
--- a/be/src/vec/exec/jni_connector.cpp
+++ b/be/src/vec/exec/jni_connector.cpp
@@ -94,13 +94,16 @@ Status JniConnector::open(RuntimeState* state, RuntimeProfile* profile) {
Status JniConnector::init(
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range) {
- _generate_predicates(colname_to_value_range);
- if (_predicates_length != 0 && _predicates != nullptr) {
- int64_t predicates_address = (int64_t)_predicates.get();
- // We can call org.apache.doris.common.jni.vec.ScanPredicate#parseScanPredicates to parse the
- // serialized predicates in java side.
- _scanner_params.emplace("push_down_predicates", std::to_string(predicates_address));
- }
+ // TODO: This logic need to be changed.
+ // See the comment of "predicates" field in JniScanner.java
+
+ // _generate_predicates(colname_to_value_range);
+ // if (_predicates_length != 0 && _predicates != nullptr) {
+ // int64_t predicates_address = (int64_t)_predicates.get();
+ // // We can call org.apache.doris.common.jni.vec.ScanPredicate#parseScanPredicates to parse the
+ // // serialized predicates in java side.
+ // _scanner_params.emplace("push_down_predicates", std::to_string(predicates_address));
+ // }
return Status::OK();
}
diff --git a/be/src/vec/exec/jni_connector.h b/be/src/vec/exec/jni_connector.h
index ccc36021681..c053c8a1f61 100644
--- a/be/src/vec/exec/jni_connector.h
+++ b/be/src/vec/exec/jni_connector.h
@@ -139,6 +139,9 @@ public:
char_ptr += s->size;
}
} else {
+ // FIXME: it can not handle decimal type correctly.
+ // but this logic is deprecated and not used.
+ // so may be deleted or fixed later.
for (const CppType* v : values) {
int type_len = sizeof(CppType);
*reinterpret_cast<int*>(char_ptr) = type_len;
diff --git a/fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroJNIScanner.java b/fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroJNIScanner.java
index 75cbc721e31..85e12d7c86b 100644
--- a/fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroJNIScanner.java
+++ b/fe/be-java-extensions/avro-scanner/src/main/java/org/apache/doris/avro/AvroJNIScanner.java
@@ -19,7 +19,6 @@ package org.apache.doris.avro;
import org.apache.doris.common.jni.JniScanner;
import org.apache.doris.common.jni.vec.ColumnType;
-import org.apache.doris.common.jni.vec.ScanPredicate;
import org.apache.doris.common.jni.vec.TableSchema;
import org.apache.doris.thrift.TFileType;
@@ -152,7 +151,7 @@ public class AvroJNIScanner extends JniScanner {
}
this.avroReader.open(new Configuration());
if (!isGetTableSchema) {
- initTableInfo(requiredTypes, requiredFields, new ScanPredicate[0], fetchSize);
+ initTableInfo(requiredTypes, requiredFields, fetchSize);
}
}
diff --git a/fe/be-java-extensions/hudi-scanner/src/main/java/org/apache/doris/hudi/HudiJniScanner.java b/fe/be-java-extensions/hudi-scanner/src/main/java/org/apache/doris/hudi/HudiJniScanner.java
index daf8d4a21fa..7044915eeb5 100644
--- a/fe/be-java-extensions/hudi-scanner/src/main/java/org/apache/doris/hudi/HudiJniScanner.java
+++ b/fe/be-java-extensions/hudi-scanner/src/main/java/org/apache/doris/hudi/HudiJniScanner.java
@@ -20,7 +20,6 @@ package org.apache.doris.hudi;
import org.apache.doris.common.jni.JniScanner;
import org.apache.doris.common.jni.vec.ColumnType;
-import org.apache.doris.common.jni.vec.ScanPredicate;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.avro.generic.GenericDatumReader;
@@ -59,7 +58,6 @@ public class HudiJniScanner extends JniScanner {
private final int fetchSize;
private final String debugString;
private final HoodieSplit split;
- private final ScanPredicate[] predicates;
private final ClassLoader classLoader;
private final UserGroupInformation ugi;
@@ -124,20 +122,8 @@ public class HudiJniScanner extends JniScanner {
.collect(Collectors.joining("\n"));
try {
this.classLoader = this.getClass().getClassLoader();
- String predicatesAddressString = params.remove("push_down_predicates");
this.fetchSize = fetchSize;
this.split = new HoodieSplit(params);
- if (predicatesAddressString == null) {
- predicates = new ScanPredicate[0];
- } else {
- long predicatesAddress = Long.parseLong(predicatesAddressString);
- if (predicatesAddress != 0) {
- predicates = ScanPredicate.parseScanPredicates(predicatesAddress, split.requiredTypes());
- LOG.info("HudiJniScanner gets pushed-down predicates: " + ScanPredicate.dump(predicates));
- } else {
- predicates = new ScanPredicate[0];
- }
- }
ugi = Utils.getUserGroupInformation(split.hadoopConf());
} catch (Exception e) {
LOG.error("Failed to initialize hudi scanner, split params:\n" + debugString, e);
@@ -149,7 +135,7 @@ public class HudiJniScanner extends JniScanner {
public void open() throws IOException {
Future<?> avroFuture = avroReadPool.submit(() -> {
Thread.currentThread().setContextClassLoader(classLoader);
- initTableInfo(split.requiredTypes(), split.requiredFields(), predicates, fetchSize);
+ initTableInfo(split.requiredTypes(), split.requiredFields(), fetchSize);
long startTime = System.nanoTime();
// RecordReader will use ProcessBuilder to start a hotspot process, which may be stuck,
// so use another process to kill this stuck process.
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
index 5031a018267..07d9f77ba37 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
@@ -33,6 +33,14 @@ public abstract class JniScanner {
protected VectorTable vectorTable;
protected String[] fields;
protected ColumnType[] types;
+ @Deprecated
+ // This predicate is from BE, but no used.
+ // TODO: actually, we can generate the predicate for JNI scanner in FE's planner,
+ // then serialize it to BE, and BE pass it to JNI scanner directly.
+ // NO need to use this intermediate expression, because each JNI scanner has its
+ // own predicate expression format.
+ // For example, Paimon use "PaimonScannerUtils.decodeStringToObject(paimonPredicate)"
+ // to deserialize the predicate string to PaimonPredicate object.
protected ScanPredicate[] predicates;
protected int batchSize;
@@ -50,11 +58,9 @@ public abstract class JniScanner {
throw new UnsupportedOperationException();
}
- protected void initTableInfo(ColumnType[] requiredTypes, String[] requiredFields, ScanPredicate[] predicates,
- int batchSize) {
+ protected void initTableInfo(ColumnType[] requiredTypes, String[] requiredFields, int batchSize) {
this.types = requiredTypes;
this.fields = requiredFields;
- this.predicates = predicates;
this.batchSize = batchSize;
}
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/MockJniScanner.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/MockJniScanner.java
index bc7561e2a23..000f536e915 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/MockJniScanner.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/MockJniScanner.java
@@ -20,7 +20,6 @@ package org.apache.doris.common.jni;
import org.apache.doris.common.jni.vec.ColumnType;
import org.apache.doris.common.jni.vec.ColumnValue;
-import org.apache.doris.common.jni.vec.ScanPredicate;
import org.apache.log4j.Logger;
@@ -187,15 +186,7 @@ public class MockJniScanner extends JniScanner {
for (int i = 0; i < types.length; i++) {
columnTypes[i] = ColumnType.parseType(requiredFields[i], types[i]);
}
- ScanPredicate[] predicates = new ScanPredicate[0];
- if (params.containsKey("push_down_predicates")) {
- long predicatesAddress = Long.parseLong(params.get("push_down_predicates"));
- if (predicatesAddress != 0) {
- predicates = ScanPredicate.parseScanPredicates(predicatesAddress, columnTypes);
- LOG.info("MockJniScanner gets pushed-down predicates: " + ScanPredicate.dump(predicates));
- }
- }
- initTableInfo(columnTypes, requiredFields, predicates, batchSize);
+ initTableInfo(columnTypes, requiredFields, batchSize);
}
@Override
diff --git a/fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java b/fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
index 95bf748658b..1fdbd585e19 100644
--- a/fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
+++ b/fe/be-java-extensions/max-compute-scanner/src/main/java/org/apache/doris/maxcompute/MaxComputeJniScanner.java
@@ -19,7 +19,6 @@ package org.apache.doris.maxcompute;
import org.apache.doris.common.jni.JniScanner;
import org.apache.doris.common.jni.vec.ColumnType;
-import org.apache.doris.common.jni.vec.ScanPredicate;
import com.aliyun.odps.Column;
import com.aliyun.odps.OdpsType;
@@ -99,15 +98,7 @@ public class MaxComputeJniScanner extends JniScanner {
for (int i = 0; i < types.length; i++) {
columnTypes[i] = ColumnType.parseType(requiredFields[i], types[i]);
}
- ScanPredicate[] predicates = new ScanPredicate[0];
- if (params.containsKey("push_down_predicates")) {
- long predicatesAddress = Long.parseLong(params.get("push_down_predicates"));
- if (predicatesAddress != 0) {
- predicates = ScanPredicate.parseScanPredicates(predicatesAddress, columnTypes);
- LOG.info("MaxComputeJniScanner gets pushed-down predicates: " + ScanPredicate.dump(predicates));
- }
- }
- initTableInfo(columnTypes, requiredFields, predicates, batchSize);
+ initTableInfo(columnTypes, requiredFields, batchSize);
}
public void refreshTableScan() {
@@ -133,9 +124,8 @@ public class MaxComputeJniScanner extends JniScanner {
}
@Override
- protected void initTableInfo(ColumnType[] requiredTypes, String[] requiredFields, ScanPredicate[] predicates,
- int batchSize) {
- super.initTableInfo(requiredTypes, requiredFields, predicates, batchSize);
+ protected void initTableInfo(ColumnType[] requiredTypes, String[] requiredFields, int batchSize) {
+ super.initTableInfo(requiredTypes, requiredFields, batchSize);
readColumns = new ArrayList<>();
readColumnsToId = new HashMap<>();
for (int i = 0; i < fields.length; i++) {
diff --git a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
index fce74a27dc6..c1c6efd2b79 100644
--- a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
+++ b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
@@ -19,7 +19,6 @@ package org.apache.doris.paimon;
import org.apache.doris.common.jni.JniScanner;
import org.apache.doris.common.jni.vec.ColumnType;
-import org.apache.doris.common.jni.vec.ScanPredicate;
import org.apache.doris.common.jni.vec.TableSchema;
import org.apache.doris.paimon.PaimonTableCache.PaimonTableCacheKey;
import org.apache.doris.paimon.PaimonTableCache.TableExt;
@@ -80,7 +79,7 @@ public class PaimonJniScanner extends JniScanner {
dbId = Long.parseLong(params.get("db_id"));
tblId = Long.parseLong(params.get("tbl_id"));
lastUpdateTime = Long.parseLong(params.get("last_update_time"));
- initTableInfo(columnTypes, requiredFields, new ScanPredicate[0], batchSize);
+ initTableInfo(columnTypes, requiredFields, batchSize);
paimonOptionParams = params.entrySet().stream()
.filter(kv -> kv.getKey().startsWith(PAIMON_OPTION_PREFIX))
.collect(Collectors
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org