Skip to content

Commit abfbc18

Browse files
committed
[hotfix-36643][kudu] kudu维表空指针异常,修改查询谓词逻辑
1 parent f67993f commit abfbc18

File tree

1 file changed

+13
-1
lines changed

1 file changed

+13
-1
lines changed

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,12 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, BaseRow input, Re
161161
// @wenbaoup fix bug
162162
inputParams.forEach((key, value) -> {
163163
Object transformValue = transformValue(value);
164+
if (transformValue == null) {
165+
scannerBuilder.addPredicate(
166+
KuduPredicate.newIsNullPredicate(schema.getColumn(key))
167+
);
168+
return;
169+
}
164170
scannerBuilder.addPredicate(
165171
KuduPredicate.newInListPredicate(
166172
schema.getColumn(key),
@@ -228,7 +234,7 @@ private Object transformValue(Object value) {
228234
public String buildCacheKey(Map<String, Object> inputParams) {
229235
StringBuilder sb = new StringBuilder();
230236
for (Object ele : inputParams.values()) {
231-
sb.append(ele.toString())
237+
sb.append(ele)
232238
.append("_");
233239
}
234240

@@ -295,6 +301,11 @@ public GetListRowCB() {
295301

296302
@Override
297303
public Deferred<List<Row>> call(RowResultIterator results) throws Exception {
304+
if (results == null) {
305+
resultFuture.complete(Collections.emptyList());
306+
return null;
307+
}
308+
298309
for (RowResult result : results) {
299310
Map<String, Object> oneRow = Maps.newHashMap();
300311
for (String sideFieldName1 : StringUtils.split(sideInfo.getSideSelectFields(), ",")) {
@@ -327,6 +338,7 @@ public Deferred<List<Row>> call(RowResultIterator results) throws Exception {
327338
}
328339
}
329340

341+
resultFuture.complete(Collections.emptyList());
330342
return null;
331343
}
332344
}

0 commit comments

Comments
 (0)