This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 4f900d07f6b [opt](lance) push down LIMIT into Lance fragment scanners
(#66608)
4f900d07f6b is described below
commit 4f900d07f6bbe9da05dea5dbf5becb75fe0852ee
Author: jay <[email protected]>
AuthorDate: Fri Aug 14 10:05:27 2026 +0800
[opt](lance) push down LIMIT into Lance fragment scanners (#66608)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #65730, #66581
Problem Summary:
Ordinary Lance scans currently read **every row of a fragment** even
when the
query only needs the first N rows (e.g. `SELECT ... LIMIT 10`). Lance
applies
its own LIMIT *after* the scanner's filter, so the query LIMIT can be
forwarded
to each fragment scanner and let it stop early, cutting IO and decode
cost.
**How it is fixed**
- `thrift`: add an optional `TLanceFileDesc.limit`.
- `FE` (`LanceScanNode`): push the query limit into each fragment split
via
`canPushDownLimit()`, and surface `lanceLimit` in the explain output.
- `BE` (`lance_reader`): forward it to the scanner through
`lance_scanner_set_limit` for ordinary scans; vector search keeps its
own
`top_k` limit.
**Correctness**
The limit is pushed **only when all predicates are already pushed into
Lance**
(no residual Doris conjunct). Otherwise Doris still re-filters the
returned rows,
and truncating a fragment early could drop valid results.
`OFFSET` needs no special handling: Nereids' `SplitLimit` rewrites
`Limit(limit, offset)` into a global `Limit(limit, offset)` over a local
`Limit(limit + offset, 0)`, and that local bound is what reaches the
scan node.
So `getLimit()` already includes the offset; each fragment fetches up to
`limit + offset` rows and the upper global LIMIT still applies the
offset and
the final bound. Per-fragment truncation is therefore always safe.
**Behavior change**
Query results are unchanged. Only the number of rows scanned per
fragment is
reduced for LIMIT queries; the explain output shows an extra
`lanceLimit=N`
line when the limit is pushed.
### Release note
Push down LIMIT into Lance fragment scanners to reduce the rows scanned
for
`LIMIT` / `LIMIT ... OFFSET` queries over Lance tables.
### Check List (For Author)
- Test
- [x] Unit Test (`LanceThriftContractTest` covers the limit round-trip
and the no-limit case)
- [ ] Manual test — `SELECT * FROM <lance_tbl> LIMIT 10` returns 10 rows
and `EXPLAIN` shows `lanceLimit=10`; a query with a non-pushable
predicate keeps the limit out of the scan
- Behavior changed:
- [x] No.
- Does this need documentation?
- [x] No.
---
be/src/format_v2/table/lance_reader.cpp | 8 +++++++
.../datasource/lance/source/LanceScanNode.java | 22 +++++++++++++++++++
.../doris/datasource/LanceThriftContractTest.java | 25 +++++++++++++++++++++-
gensrc/thrift/PlanNodes.thrift | 4 ++++
4 files changed, 58 insertions(+), 1 deletion(-)
diff --git a/be/src/format_v2/table/lance_reader.cpp
b/be/src/format_v2/table/lance_reader.cpp
index 4f1229901a9..142c35ac4d4 100644
--- a/be/src/format_v2/table/lance_reader.cpp
+++ b/be/src/format_v2/table/lance_reader.cpp
@@ -683,6 +683,14 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
return _lance_error("set Lance scanner fragment ids");
}
}
+ // Ordinary scans may carry a pushed-down LIMIT. The FE only sets it when
all predicates are
+ // pushed into Lance, so the scanner can safely stop after `limit` rows.
Vector search manages
+ // its own top_k limit in _configure_vector_search, so skip it here.
+ if (!_vector_search && lance_params.__isset.limit && lance_params.limit >
0) {
+ if (lance_scanner_set_limit(scanner, lance_params.limit) != 0) {
+ return _lance_error("set Lance scanner limit");
+ }
+ }
if (_vector_search) {
RETURN_IF_ERROR(_configure_vector_search(scanner));
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
index 7c363ab6e59..4c433a30bce 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
@@ -114,6 +114,19 @@ public class LanceScanNode extends FileQueryScanNode {
}
}
+ // A fragment-level LIMIT can be pushed into an ordinary Lance scan only
when every predicate
+ // is already pushed into Lance (conjuncts is empty). Otherwise Doris
re-filters the returned
+ // rows and truncating a fragment early could drop valid results.
+ //
+ // OFFSET needs no special handling: the Nereids SplitLimit rule rewrites
Limit(limit, offset)
+ // into a global Limit(limit, offset) over a local Limit(limit + offset,
0), and the local
+ // bound is what lands on this scan node. So getLimit() already accounts
for the offset and
+ // getOffset() is always 0 here; each fragment fetches up to limit +
offset rows and the upper
+ // global LIMIT still applies the offset and the final bound.
+ private boolean canPushDownLimit() {
+ return hasLimit() && conjuncts.isEmpty();
+ }
+
@Override
protected void convertPredicate() {
if (isExternalSearch()) {
@@ -199,6 +212,12 @@ public class LanceScanNode extends FileQueryScanNode {
"Ordinary Lance scan split must contain one fragment");
}
lanceParams.setFragmentIds(Collections.singletonList(lanceSplit.getFragmentId()));
+ // Push LIMIT into each fragment scanner only when it is safe to
truncate a single
+ // fragment early. See canPushDownLimit(). Each scanner still
returns at most `limit`
+ // rows and the upper LIMIT operator enforces the final bound
across fragments.
+ if (canPushDownLimit()) {
+ lanceParams.setLimit(getLimit());
+ }
}
TTableFormatFileDesc tableFormatParams = new TTableFormatFileDesc();
@@ -253,6 +272,9 @@ public class LanceScanNode extends FileQueryScanNode {
.append(((LanceExternalCatalog)
lanceTable.getCatalog()).getLanceCatalogType()).append("\n");
result.append(prefix).append("lanceVersion=").append(plannedVersion).append("\n");
result.append(prefix).append("lanceFragments=").append(plannedFragments).append("\n");
+ if (canPushDownLimit()) {
+
result.append(prefix).append("lanceLimit=").append(getLimit()).append("\n");
+ }
if (!lancePushdownPredicate.isEmpty()) {
result.append(prefix).append("lancePushdownPredicate=")
.append(lancePushdownPredicate).append("\n");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
index e28dd3c7382..dc21fafa314 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
@@ -36,7 +36,8 @@ public class LanceThriftContractTest {
TLanceFileDesc lanceDesc = new TLanceFileDesc()
.setDatasetUri("s3://warehouse/db/table.lance")
.setFragmentIds(Arrays.asList(7L, 11L))
- .setVersion(42L);
+ .setVersion(42L)
+ .setLimit(100L);
TTableFormatFileDesc source = new TTableFormatFileDesc()
.setTableFormatType(TableFormatType.LANCE.value())
.setLanceParams(lanceDesc);
@@ -53,5 +54,27 @@ public class LanceThriftContractTest {
Assert.assertEquals("s3://warehouse/db/table.lance",
restored.getLanceParams().getDatasetUri());
Assert.assertEquals(Arrays.asList(7L, 11L),
restored.getLanceParams().getFragmentIds());
Assert.assertEquals(42L, restored.getLanceParams().getVersion());
+ Assert.assertTrue(restored.getLanceParams().isSetLimit());
+ Assert.assertEquals(100L, restored.getLanceParams().getLimit());
+ }
+
+ @Test
+ public void testLanceDescriptorWithoutLimit() throws Exception {
+ TLanceFileDesc lanceDesc = new TLanceFileDesc()
+ .setDatasetUri("s3://warehouse/db/table.lance")
+ .setFragmentIds(Arrays.asList(1L))
+ .setVersion(1L);
+ TTableFormatFileDesc source = new TTableFormatFileDesc()
+ .setTableFormatType(TableFormatType.LANCE.value())
+ .setLanceParams(lanceDesc);
+
+ TSerializer serializer = new TSerializer(new
TCompactProtocol.Factory());
+ byte[] bytes = serializer.serialize(source);
+
+ TTableFormatFileDesc restored = new TTableFormatFileDesc();
+ new TDeserializer(new
TCompactProtocol.Factory()).deserialize(restored, bytes);
+
+ // A scan without a pushable LIMIT must leave the field unset so the
BE reads all rows.
+ Assert.assertFalse(restored.getLanceParams().isSetLimit());
}
}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 3b7c377110a..d3f0583c220 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -527,6 +527,10 @@ struct TLanceFileDesc {
1: optional string dataset_uri
2: optional list<i64> fragment_ids
3: optional i64 version
+ // Per-split row limit pushed down from the query LIMIT. Each scanner
returns at
+ // most this many rows; the upper LIMIT operator still enforces the global
bound.
+ // Only set for ordinary scans whose predicates are fully pushed into
Lance.
+ 4: optional i64 limit
}
struct TTableFormatFileDesc {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
