Expose filter_join_indices in Java - #24097
Conversation
|
Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually. Contributors can view more details about this message here. |
bdice
added
improvement
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (4)
Included review availability: Your plan provides up to 12 included reviews per hour; 11 remain after this review. 📝 SummarySummary by CodeRabbit
WalkthroughAdds the ChangesFilter join gather maps
Estimated code review effort: 3 (Moderate) | ~30 minutes Suggested reviewers: Merge Risk: 🟡 Moderate · up to The new Java filtering API depends on the JNI/native join contract matching the Java join-kind values exactly. Until that mapping and call signature are confirmed, filtered joins could return incorrect results, so this should be resolved before merge. 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
Full details: Linked Issues checkExplanation The implementation meets the binding, argument, ownership, and join-kind requirements in [ Resolution Add tests for duplicate equality keys, a left row where all conditional matches fail, and nullable predicates. Then confirm the implementation preserves input gather maps and returns newly allocated output maps for these cases as required by [
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
| /** Retain every left row, using an invalid right index when no pair satisfies the condition. */ | ||
| LEFT(1), | ||
| /** Retain every row from both sides, splitting row pairs that do not satisfy the condition. */ | ||
| FULL(2); |
There was a problem hiding this comment.
JoinKind.FULL currently produces extra unmatched rows when duplicate equality keys have a mixture of passing and failing predicate candidates. I reproduced this with this PR's CUDA 12 CI JAR, whose embedded revision is e06741dfed37ff61eefc40390a518c9ea4e8595d.
For left (key=1, value=10), right [(1,5), (1,15)], and left.key == right.key && left.value > right.value, the expected index pairs are (0,0), (null,1). The filter also emits (0,null), although that left row already has a successful match. If both candidates fail, it emits the same unmatched left row twice. Duplicating the other side or using a null-valued predicate candidate exposes the same problem.
The existing conditionalFullJoinGatherMaps API agrees with the independent host oracle on these cases. The underlying libcudf FULL branch splits each failed pair independently, without checking whether the source row has another passing match. That implementation predates this PR, but exposing it here makes the new FULL binding unsafe for general full joins. Could we fix it upstream and add duplicate-key coverage, or leave FULL out of this binding until that fix is available? Deduplicating null-extended rows alone would not fix the mixed pass/fail case.
Observed output (null denotes Integer.MIN_VALUE in the gather map):
full_duplicate_mixed MISMATCH expected=[(0,0), (null,1)] actual=[(0,0), (0,null), (null,1)]
full_duplicate_all_rejected MISMATCH expected=[(0,null), (null,0), (null,1)] actual=[(0,null), (0,null), (null,0), (null,1)]
full_duplicate_left MISMATCH expected=[(0,0), (1,null)] actual=[(0,0), (1,null), (null,0)]
full_nullable_duplicates MISMATCH expected=[(0,0), (null,1)] actual=[(0,0), (0,null), (null,1)]
Runnable reproducer, including controls and the input-map contract probes in my other comment
Save as FilterJoinProbe.java. Compile and run with this PR's cuDF JAR plus SLF4J API/binding on the classpath, with the CUDA runtime libraries available:
javac -cp "$CUDF_JAR" FilterJoinProbe.java
java -ea -cp ".:$CUDF_JAR:$SLF4J_API_JAR:$SLF4J_SIMPLE_JAR" FilterJoinProbeThe program intentionally exits 1 when it reports mismatches. In my run, seven controls passed, four FULL cases failed, and the two cross-kind input-map probes exposed the contract behavior described separately. All five conditional-full-join reference checks agreed with the host oracle. The probe also verifies unchanged input contents and reads each output after closing its input maps.
import ai.rapids.cudf.*;
import ai.rapids.cudf.ast.*;
import java.util.*;
public class FilterJoinProbe {
static final int INV = Integer.MIN_VALUE;
static int mismatches = 0;
static List<String> read(GatherMap[] maps) {
int n = Math.toIntExact(maps[0].getRowCount());
List<String> out = new ArrayList<>();
if (n == 0) return out;
try (ColumnView l = maps[0].toColumnView(0, n);
ColumnView r = maps[1].toColumnView(0, n);
HostColumnVector lh = l.copyToHost(); HostColumnVector rh = r.copyToHost()) {
for (int i = 0; i < n; i++) out.add(pair(lh.getInt(i), rh.getInt(i)));
}
Collections.sort(out);
return out;
}
static String pair(int l, int r) {
return "(" + (l == INV ? "null" : l) + "," + (r == INV ? "null" : r) + ")";
}
static List<String> oracle(JoinKind kind, Integer[] lk, Integer[] lv,
Integer[] rk, Integer[] rv) {
List<String> out = new ArrayList<>();
boolean[] matchedRight = new boolean[rk.length];
for (int l = 0; l < lk.length; l++) {
boolean matched = false;
for (int r = 0; r < rk.length; r++) {
if (lk[l] != null && rk[r] != null && lk[l].equals(rk[r]) &&
lv[l] != null && rv[r] != null && lv[l] > rv[r]) {
matched = true;
matchedRight[r] = true;
out.add(pair(l, r));
}
}
if (!matched && kind != JoinKind.INNER) out.add(pair(l, INV));
}
if (kind == JoinKind.FULL) {
for (int r = 0; r < rk.length; r++) if (!matchedRight[r]) out.add(pair(INV, r));
}
Collections.sort(out);
return out;
}
static Table table(Integer[] v) {
try (ColumnVector c = ColumnVector.fromBoxedInts(v)) { return new Table(c); }
}
static GatherMap[] equality(JoinKind kind, Table keys, HashJoin hash) {
switch (kind) {
case INNER: return keys.innerJoinGatherMaps(hash);
case LEFT: return keys.leftJoinGatherMaps(hash);
case FULL: return keys.fullJoinGatherMaps(hash);
default: throw new AssertionError(kind);
}
}
static List<String> fullReference(Integer[] lk, Integer[] lv, Integer[] rk, Integer[] rv) {
BinaryOperation expr = new BinaryOperation(BinaryOperator.LOGICAL_AND,
new BinaryOperation(BinaryOperator.EQUAL,
new ColumnReference(0, TableReference.LEFT), new ColumnReference(0, TableReference.RIGHT)),
new BinaryOperation(BinaryOperator.GREATER,
new ColumnReference(1, TableReference.LEFT), new ColumnReference(1, TableReference.RIGHT)));
try (ColumnVector lkc = ColumnVector.fromBoxedInts(lk);
ColumnVector lvc = ColumnVector.fromBoxedInts(lv);
ColumnVector rkc = ColumnVector.fromBoxedInts(rk);
ColumnVector rvc = ColumnVector.fromBoxedInts(rv);
Table left = new Table(lkc, lvc); Table right = new Table(rkc, rvc);
CompiledExpression condition = expr.compile()) {
GatherMap[] maps = left.conditionalFullJoinGatherMaps(right, condition);
try (GatherMap l = maps[0]; GatherMap r = maps[1]) { return read(maps); }
}
}
static void run(String name, JoinKind seed, JoinKind kind,
Integer[] lk, Integer[] lv, Integer[] rk, Integer[] rv) {
List<String> expected = oracle(kind, lk, lv, rk, rv);
if (kind == JoinKind.FULL) {
List<String> reference = fullReference(lk, lv, rk, rv);
if (!expected.equals(reference)) throw new AssertionError("Native reference differs from CPU oracle");
System.out.println(name + " conditionalFullJoinGatherMaps_reference=PASS " + reference);
}
BinaryOperation expr = new BinaryOperation(BinaryOperator.GREATER,
new ColumnReference(0, TableReference.LEFT), new ColumnReference(0, TableReference.RIGHT));
try (Table leftKeys = table(lk); Table leftValues = table(lv);
Table rightKeys = table(rk); Table rightValues = table(rv);
HashJoin hash = new HashJoin(rightKeys, false); CompiledExpression condition = expr.compile()) {
GatherMap[] filtered;
GatherMap[] input = equality(seed, leftKeys, hash);
try (GatherMap l = input[0]; GatherMap r = input[1]) {
List<String> before = read(input);
filtered = Table.filterJoinGatherMaps(l, r, leftValues, rightValues, condition, kind);
if (!before.equals(read(input))) throw new AssertionError("Input mutated: " + name);
}
// Read after input closure: the output must own independent buffers.
try (GatherMap l = filtered[0]; GatherMap r = filtered[1]) {
List<String> actual = read(filtered);
boolean equal = expected.equals(actual);
if (!equal) mismatches++;
System.out.println(name + " " + (equal ? "PASS" : "MISMATCH") +
" expected=" + expected + " actual=" + actual);
}
}
}
public static void main(String[] args) {
System.out.println("Table loaded from " + Table.class.getProtectionDomain().getCodeSource().getLocation());
run("inner_duplicates_control", JoinKind.INNER, JoinKind.INNER,
new Integer[]{1}, new Integer[]{10}, new Integer[]{1,1}, new Integer[]{5,15});
run("left_duplicates_control", JoinKind.LEFT, JoinKind.LEFT,
new Integer[]{1}, new Integer[]{10}, new Integer[]{1,1}, new Integer[]{5,15});
run("full_unique_control", JoinKind.FULL, JoinKind.FULL,
new Integer[]{1,2}, new Integer[]{10,20}, new Integer[]{1,3}, new Integer[]{15,30});
run("left_all_rejected_control", JoinKind.LEFT, JoinKind.LEFT,
new Integer[]{1}, new Integer[]{0}, new Integer[]{1,1}, new Integer[]{5,15});
run("left_nullable_control", JoinKind.LEFT, JoinKind.LEFT,
new Integer[]{1,2,3,1}, new Integer[]{10,20,30,null},
new Integer[]{1,1,2}, new Integer[]{5,15,25});
run("full_duplicate_mixed", JoinKind.FULL, JoinKind.FULL,
new Integer[]{1}, new Integer[]{10}, new Integer[]{1,1}, new Integer[]{5,15});
run("full_duplicate_all_rejected", JoinKind.FULL, JoinKind.FULL,
new Integer[]{1}, new Integer[]{0}, new Integer[]{1,1}, new Integer[]{5,15});
run("full_duplicate_left", JoinKind.FULL, JoinKind.FULL,
new Integer[]{1,1}, new Integer[]{10,0}, new Integer[]{1}, new Integer[]{5});
run("full_nullable_duplicates", JoinKind.FULL, JoinKind.FULL,
new Integer[]{1}, new Integer[]{10}, new Integer[]{1,1}, new Integer[]{5,null});
run("left_empty_right_control", JoinKind.LEFT, JoinKind.LEFT,
new Integer[]{1}, new Integer[]{10}, new Integer[]{}, new Integer[]{});
run("inner_empty_control", JoinKind.INNER, JoinKind.INNER,
new Integer[]{1}, new Integer[]{10}, new Integer[]{2}, new Integer[]{5});
run("inner_from_left_maps_contract", JoinKind.LEFT, JoinKind.INNER,
new Integer[]{1}, new Integer[]{10}, new Integer[]{2}, new Integer[]{5});
run("left_from_empty_inner_maps_contract", JoinKind.INNER, JoinKind.LEFT,
new Integer[]{1}, new Integer[]{10}, new Integer[]{2}, new Integer[]{5});
System.out.println("SQL_ORACLE_MISMATCHES=" + mismatches);
if (mismatches != 0) System.exit(1);
}
}
There was a problem hiding this comment.
Thank you for taking this change for a spin, @wjxiz1992. This is a great catch.
This looks like a bug in libcudf, and not quite the JNI connective tissue.
| * @param leftTable left table containing the columns referenced by the condition | ||
| * @param rightTable right table containing the columns referenced by the condition | ||
| * @param condition conditional expression to evaluate for each pair | ||
| * @param joinKind join semantics to apply when the condition does not match |
There was a problem hiding this comment.
Could the Javadoc specify which input-map kinds are valid for each JoinKind, including unmatched-sentinel and empty-map behavior? The current wording makes the enum look like it selects the desired output join semantics independently of the input maps.
Using this PR's exact-head CUDA 12 JAR, I observed:
- LEFT equality maps containing
(0, INT_MIN)passed withJoinKind.INNERretain that unmatched pair, despite INNER being documented to retain only pairs satisfying the condition. - Empty INNER equality maps from two nonempty, disjoint tables passed with
JoinKind.LEFTreturn empty, rather than an unmatched left row.
These are input-contract cases, not failures of correctly paired INNER/LEFT calls. If the intended contract requires maps from the corresponding join kind, documenting that restriction and its sentinel behavior would make this much safer to compose. Alternatively, unsupported combinations could be rejected or normalized. This matters for cudf-spark's existing inner-join -> predicate-filter -> outer-completion pipeline.
The complete runnable reproducer includes these two calls (using its run helper):
run("inner_from_left_maps_contract", JoinKind.LEFT, JoinKind.INNER,
new Integer[]{1}, new Integer[]{10}, new Integer[]{2}, new Integer[]{5});
run("left_from_empty_inner_maps_contract", JoinKind.INNER, JoinKind.LEFT,
new Integer[]{1}, new Integer[]{10}, new Integer[]{2}, new Integer[]{5});Observed output; expected is the relational result for the requested output join kind:
inner_from_left_maps_contract MISMATCH expected=[] actual=[(0,null)]
left_from_empty_inner_maps_contract MISMATCH expected=[(0,null)] actual=[]
4 participants
Description
Adds a Java/JNI binding for
cudf::filter_join_indices, including Java join-kind selection and ownership-preserving conversion betweenGatherMapinputs and outputs. This lets Java callers compose reusableHashJoinequality gather maps with AST predicate filtering for inner, left, and full joins.Closes #24096.
Checklist