Skip to content

Commit 531fcbd

Browse files
Gabriel39Gabriel39
committed
[fix](lance) Preserve rewritten label pushdown and fragment parallelism
Adopt merged upstream lance-c main and retain only the unchanged Foyer cache patch until its upstream PR merges. Convert non-null string overlap needles, keep complement scans split by fragment, and verify committed fixture coverage and runtime partial-index and scanner-limit paths. Validation: 61 focused FE tests, Checkstyle, 451 Rust/C API tests, 27 FE Substrait integration scans, fixture checks, and patch harnesses. Full Doris SQL regression on this revision is pending CI.
1 parent f46234d commit 531fcbd

9 files changed

Lines changed: 203 additions & 37 deletions

File tree

‎docker/thirdparties/docker-compose/iceberg/scripts/lance_build_array_predicates.py‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,9 +53,37 @@ def build(output):
5353
dataset.create_scalar_index("category", "BTREE", name="category_idx")
5454
if name == "partial":
5555
lance.write_dataset(table.slice(8), str(path), mode="append", max_rows_per_file=8)
56+
check(output)
57+
58+
59+
def check(output):
60+
if not __debug__:
61+
raise RuntimeError("Fixture verification requires assertions; do not use python -O")
62+
table = pa.table({"id": pa.array(range(16), type=pa.int64()),
63+
"labels": pa.array(LABELS, type=pa.list_(pa.string())),
64+
"category": pa.array([i % 3 for i in range(16)], type=pa.int32())})
65+
for name in ["indexed", "partial", "unindexed"]:
66+
path = output / (name + ".lance")
5667
dataset = lance.dataset(str(path))
5768
assert dataset.to_table().to_pydict() == table.to_pydict()
58-
assert len(dataset.get_fragments()) == 2
69+
fragments = dataset.get_fragments()
70+
assert len(fragments) == 2
71+
assert [f.to_table(columns=["id"])["id"].to_pylist() for f in fragments] == [
72+
list(range(8)), list(range(8, 16))]
73+
indexes = {index["name"]: index for index in dataset.list_indices()}
74+
assert set(indexes) == (set() if name == "unindexed" else {"labels_idx", "category_idx"})
75+
expected_coverage = {f.fragment_id for f in (fragments[:1] if name == "partial" else fragments)}
76+
# Names alone do not prove the mixed indexed/unindexed fixture: verify each
77+
# physical domain so rebuilding both fragments into an index cannot pass silently.
78+
for index_name, field, index_type in [("labels_idx", "labels", "LabelList"),
79+
("category_idx", "category", "BTree")]:
80+
if name == "unindexed":
81+
continue
82+
index = indexes[index_name]
83+
assert index["fields"] == [field] and index["type"] == index_type
84+
assert index["fragment_ids"] == expected_coverage, (name, index_name, index)
85+
if name == "partial":
86+
assert len({f.fragment_id for f in fragments} - expected_coverage) == 1
5987
for predicate, expected in [
6088
("array_contains(labels, 'red')", [0, 2, 5, 7, 8, 10, 13, 15]),
6189
("array_contains(labels, 'red') AND array_contains(labels, 'blue')", [2, 7, 10, 15]),
@@ -73,4 +101,6 @@ def build(output):
73101
parser = argparse.ArgumentParser(description=__doc__)
74102
parser.add_argument("--output", type=Path, default=Path(__file__).parent /
75103
"preinstalled_data/lance/predicate_arrays")
76-
build(parser.parse_args().output)
104+
parser.add_argument("--check", action="store_true", help="Verify committed fixtures without rebuilding")
105+
args = parser.parse_args()
106+
(check if args.check else build)(args.output)

‎docker/thirdparties/docker-compose/iceberg/scripts/lance_build_preinstalled_catalog.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@
8989
import lance_namespace
9090
import pyarrow as pa
9191
import pyarrow.ipc as ipc
92-
from lance_build_array_predicates import build as build_array_predicates
92+
from lance_build_array_predicates import build as build_array_predicates, check as check_array_predicates
9393
from lance_build_multivector import build as build_multivector, check as check_multivector
9494
from lance_build_nested_null import build as build_nested_null, check as check_nested_null
9595
from lance_build_time_travel import check as check_time_travel
@@ -1672,6 +1672,8 @@ def check_fts_dataset(location: str, *, table_name: str, index_name: str,
16721672

16731673

16741674
def check_catalog(root: Path) -> None:
1675+
# The SQL pushdown suite also relies on these independently published datasets.
1676+
check_array_predicates(root / "predicate_arrays")
16751677
check_data_shapes()
16761678
namespace = lance_namespace.connect("dir", {"root": str(root)})
16771679
tables = namespace.list_tables(ListTablesRequest(id=[NAMESPACE]))

‎fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LancePredicateConverter.java‎

Lines changed: 44 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
package org.apache.doris.datasource.lance.source;
1919

20+
import org.apache.doris.analysis.ArrayLiteral;
2021
import org.apache.doris.analysis.BinaryPredicate;
2122
import org.apache.doris.analysis.BoolLiteral;
2223
import org.apache.doris.analysis.CompoundPredicate;
@@ -275,6 +276,8 @@ private Optional<Expression> convertFunction(FunctionCallExpr function) {
275276
}
276277
String functionName = function.getFnName().getFunction().toLowerCase(Locale.ROOT);
277278
switch (functionName) {
279+
case "arrays_overlap":
280+
return convertArraysOverlap(function);
278281
case "array_contains":
279282
return convertArrayContains(function);
280283
case "like":
@@ -291,6 +294,40 @@ private Optional<Expression> convertFunction(FunctionCallExpr function) {
291294
}
292295
}
293296

297+
private Optional<Expression> convertArraysOverlap(FunctionCallExpr function) {
298+
Expr input = function.getChild(0);
299+
Expr values = function.getChild(1);
300+
if (input instanceof ArrayLiteral) {
301+
input = function.getChild(1);
302+
values = function.getChild(0);
303+
}
304+
SlotRef slot = directSlot(input);
305+
ResolvedField field = slot == null ? null : findField(slot);
306+
if (field == null || !LanceSubstraitSerializer.supportsStringList(field.field)
307+
|| !(values instanceof ArrayLiteral) || values.getChildren().isEmpty()) {
308+
return Optional.empty();
309+
}
310+
// Nereids folds three or more membership disjuncts into arrays_overlap. Expand
311+
// only non-NULL string needles: Doris also matches NULL elements to NULL needles.
312+
List<Expression> level = new ArrayList<>();
313+
for (Expr needle : values.getChildren()) {
314+
if (!(needle instanceof StringLiteral)) {
315+
return Optional.empty();
316+
}
317+
level.add(arrayMembership(field, needle.getStringValue()));
318+
}
319+
// Keep expression depth logarithmic for large constant label lists.
320+
while (level.size() > 1) {
321+
List<Expression> next = new ArrayList<>();
322+
for (int i = 0; i < level.size(); i += 2) {
323+
next.add(i + 1 == level.size() ? level.get(i)
324+
: booleanFunction("or:bool", Arrays.asList(level.get(i), level.get(i + 1))));
325+
}
326+
level = next;
327+
}
328+
return Optional.of(level.get(0));
329+
}
330+
294331
private Optional<Expression> convertArrayContains(FunctionCallExpr function) {
295332
SlotRef slot = directSlot(function.getChild(0));
296333
ResolvedField field = slot == null ? null : findField(slot);
@@ -302,13 +339,17 @@ private Optional<Expression> convertArrayContains(FunctionCallExpr function) {
302339
// Doris matches a NULL needle against NULL elements; DataFusion does not. Also,
303340
// array_contains_all is a contiguous subsequence test, not DataFusion's set containment.
304341
// Only non-NULL scalar membership is interchangeable across the two engines.
305-
return Optional.of(Expression.ScalarFunctionInvocation.builder()
342+
return Optional.of(arrayMembership(field, needle.getStringValue()));
343+
}
344+
345+
private Expression arrayMembership(ResolvedField field, String needle) {
346+
return Expression.ScalarFunctionInvocation.builder()
306347
.declaration(EXTENSIONS.getScalarFunction(
307348
SimpleExtension.FunctionAnchor.of(LANCE_FUNCTIONS, "array_has:list_str")))
308349
.outputType(TypeCreator.of(field.field.isNullable()).BOOLEAN)
309350
.arguments(Arrays.asList(fieldReference(field),
310-
ExpressionCreator.string(false, needle.getStringValue())))
311-
.build());
351+
ExpressionCreator.string(false, needle)))
352+
.build();
312353
}
313354

314355
private Optional<Expression> convertStringPredicate(

‎fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlanner.java‎

Lines changed: 31 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,11 @@
1717

1818
package org.apache.doris.datasource.lance.source;
1919

20+
import org.apache.doris.analysis.BinaryPredicate;
2021
import org.apache.doris.analysis.CompoundPredicate;
2122
import org.apache.doris.analysis.Expr;
23+
import org.apache.doris.analysis.InPredicate;
24+
import org.apache.doris.analysis.IsNullPredicate;
2225
import org.apache.doris.analysis.SlotRef;
2326
import org.apache.doris.datasource.lance.index.LanceIndexSegmentGroup;
2427
import org.apache.doris.datasource.lance.index.LanceIndexSegmentInfo;
@@ -61,6 +64,8 @@ static Plan plan(LanceTableMetadata metadata, List<Expr> pushedConjuncts,
6164
// provides a stable winner when multiple indices cover the same number of rows.
6265
List<LanceIndexSegmentGroup> indices = new ArrayList<>(metadata.getIndexes());
6366
indices.sort(java.util.Comparator.comparing(LanceIndexSegmentGroup::getName));
67+
boolean splitComplementByFragment = pushedConjuncts.stream()
68+
.anyMatch(LanceScalarIndexPlanner::containsComplement);
6469
Plan selected = null;
6570
for (LanceIndexSegmentGroup logicalIndex : indices) {
6671
List<LanceIndexSegmentInfo> segments = logicalIndex.getSegments();
@@ -72,7 +77,8 @@ static Plan plan(LanceTableMetadata metadata, List<Expr> pushedConjuncts,
7277
|| index.getFieldIds().size() != 1 || !filterFields.contains(index.getFieldIds().get(0))) {
7378
continue;
7479
}
75-
Plan candidate = groupFragments(metadata, segments, visibleFragments);
80+
Plan candidate = groupFragments(metadata, segments, visibleFragments,
81+
splitComplementByFragment);
7682
if (candidate != null && (selected == null || candidate.coveredRows > selected.coveredRows)) {
7783
selected = candidate;
7884
}
@@ -115,8 +121,20 @@ private static Set<Integer> collectDriverFields(LanceTableMetadata metadata, Exp
115121
return fields;
116122
}
117123

124+
private static boolean containsComplement(Expr expr) {
125+
if ((expr instanceof CompoundPredicate
126+
&& ((CompoundPredicate) expr).getOp() == CompoundPredicate.Operator.NOT)
127+
|| (expr instanceof InPredicate && ((InPredicate) expr).isNotIn())
128+
|| (expr instanceof BinaryPredicate
129+
&& ((BinaryPredicate) expr).getOp() == BinaryPredicate.Operator.NE)
130+
|| (expr instanceof IsNullPredicate && ((IsNullPredicate) expr).isNotNull())) {
131+
return true;
132+
}
133+
return expr.getChildren().stream().anyMatch(LanceScalarIndexPlanner::containsComplement);
134+
}
135+
118136
private static Plan groupFragments(LanceTableMetadata metadata, List<LanceIndexSegmentInfo> segments,
119-
Map<Long, LanceFragmentInfo> visibleFragments) {
137+
Map<Long, LanceFragmentInfo> visibleFragments, boolean splitComplementByFragment) {
120138
LanceSplitBuilder splits = new LanceSplitBuilder(
121139
metadata.getDatasetUri(), metadata.getVersion(), segments.size());
122140
Set<Long> coveredFragments = new HashSet<>();
@@ -144,7 +162,17 @@ private static Plan groupFragments(LanceTableMetadata metadata, List<LanceIndexS
144162
coveredRows += Math.max(fragment.getPhysicalRows(), 0);
145163
}
146164
if (!fragments.isEmpty()) {
147-
splits.addIndexSegmentSplit(segment.getUuid(), fragments, physicalRows);
165+
// Complement selectivity is unknown. Preserve fragment-level scheduling so a
166+
// broad NOT cannot funnel all covered data through one BE. Each task retains
167+
// the same segment UUID but owns a disjoint domain, including on fallback.
168+
if (splitComplementByFragment) {
169+
for (Long fragmentId : fragments) {
170+
splits.addIndexSegmentSplit(segment.getUuid(), java.util.Collections.singletonList(fragmentId),
171+
visibleFragments.get(fragmentId).getSchedulingWeight());
172+
}
173+
} else {
174+
splits.addIndexSegmentSplit(segment.getUuid(), fragments, physicalRows);
175+
}
148176
}
149177
}
150178
return splits.isEmpty() ? null : new Plan(segments.get(0).getIndexName(), splits, coveredRows);

‎fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LancePredicateConverterTest.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
package org.apache.doris.datasource.lance;
1919

20+
import org.apache.doris.analysis.ArrayLiteral;
2021
import org.apache.doris.analysis.BinaryPredicate;
2122
import org.apache.doris.analysis.CompoundPredicate;
2223
import org.apache.doris.analysis.DateLiteral;
@@ -121,6 +122,33 @@ public void testArrayMembershipBooleanCombinations() {
121122
}
122123
}
123124

125+
@Test
126+
public void testRewrittenArrayOverlapPushdown() throws Exception {
127+
LancePredicateConverter arrays = arrayConverter(ArrowType.List.INSTANCE, ArrowType.Utf8.INSTANCE);
128+
Expr labels = new SlotRef(null, "labels");
129+
Expr values = new ArrayLiteral(ArrayType.create(Type.STRING, true),
130+
new StringLiteral("red"), new StringLiteral("blue"), new StringLiteral("green"));
131+
for (Expr predicate : Arrays.asList(arrayFunction("arrays_overlap", labels, values),
132+
arrayFunction("arrays_overlap", values, labels),
133+
new CompoundPredicate(CompoundPredicate.Operator.NOT,
134+
arrayFunction("arrays_overlap", labels, values), null))) {
135+
LancePredicateConverter.ConversionResult result = arrays.convert(Collections.singletonList(predicate));
136+
Assertions.assertTrue(result.getResidualConjuncts().isEmpty());
137+
String encoded = ExtendedExpression.parseFrom(result.getSubstraitFilter()).toString();
138+
Assertions.assertTrue(encoded.contains("array_has:list_str"));
139+
Assertions.assertTrue(encoded.contains("or:bool"));
140+
Assertions.assertTrue(encoded.contains("green"));
141+
}
142+
// A NULL needle can match a NULL element in Doris and must not become array_has.
143+
for (Expr valuesWithDifferentSemantics : Arrays.asList(
144+
new ArrayLiteral(ArrayType.create(Type.STRING, true), new StringLiteral("red"), new NullLiteral()),
145+
new ArrayLiteral(), new SlotRef(null, "labels"))) {
146+
Expr predicate = arrayFunction("arrays_overlap", labels, valuesWithDifferentSemantics);
147+
Assertions.assertEquals(Collections.singletonList(predicate),
148+
arrays.convert(Collections.singletonList(predicate)).getResidualConjuncts());
149+
}
150+
}
151+
124152
@Test
125153
public void testUnsafeArrayMembershipRemainsResidual() {
126154
LancePredicateConverter arrays = arrayConverter(ArrowType.List.INSTANCE, ArrowType.Utf8.INSTANCE);

‎fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScanNodeTest.java‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -243,11 +243,21 @@ public void testScalarSegmentSelectionSupportsSameColumnOrAndNot() throws Except
243243
setMetadata(node, scalarMetadata(Collections.singletonList(
244244
scalarSegment(UUID.randomUUID(), IndexType.BTREE, Arrays.asList(1L, 2L, 3L, 4L)))));
245245
setPushedConjuncts(node, filter);
246-
Assert.assertEquals(1, node.getSplits(20).size());
246+
int expectedSplits = ((CompoundPredicate) filter).getOp() == CompoundPredicate.Operator.NOT ? 4 : 1;
247+
Assert.assertEquals(expectedSplits, node.getSplits(20).size());
247248
setPushedConjuncts(node, new CompoundPredicate(CompoundPredicate.Operator.AND, filter, predicate));
248249
List<Split> splits = node.getSplits(20);
249-
Assert.assertEquals(1, splits.size());
250-
Assert.assertTrue(((LanceSplit) splits.get(0)).getIndexSegmentUuid().isPresent());
250+
Assert.assertEquals(expectedSplits, splits.size());
251+
List<Long> assignedFragments = new ArrayList<>();
252+
for (Split split : splits) {
253+
assignedFragments.addAll(((LanceSplit) split).getFragmentIds());
254+
Assert.assertTrue(((LanceSplit) split).getIndexSegmentUuid().isPresent());
255+
if (expectedSplits == 4) {
256+
Assert.assertEquals(1, ((LanceSplit) split).getFragmentIds().size());
257+
}
258+
}
259+
Collections.sort(assignedFragments);
260+
Assert.assertEquals(Arrays.asList(1L, 2L, 3L, 4L), assignedFragments);
251261
}
252262
}
253263

0 commit comments

Comments
 (0)