Skip to content

Commit 138196f

Browse files
beinanBeinan Wang
beinan
and
Beinan Wang
committed
[Feature] Add Lance connector scan node and plan fragment builder
Add the execution-planning layer for the Lance connector: - LanceScanNode: builds scan ranges with Lance dataset URI, split info, and JNI reader flag, populating THdfsScanRange lance fields - PlanFragmentBuilder: visitPhysicalLanceScan wires predicates, min/max expressions, and scan range locations into LanceScanNode - LanceTable.isSupported() returns true to allow query planning - LancePlanTest: unit test for explain plan output - ProfilingExecPlanTest: add LANCE_SCAN to known node name set - LanceMetadataTest: update assertion to assertTrue(isSupported) Signed-off-by: Beinan Wang <soros.wang@yahoo.com> Co-Authored-By: Beinan Wang <beinanwang@microsoft.com> Signed-off-by: Beinan Wang <soros.wang@yahoo.com>
1 parent 601ce03 commit 138196f

6 files changed

Lines changed: 367 additions & 3 deletions

File tree

‎fe/fe-core/src/main/java/com/starrocks/catalog/LanceTable.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,6 @@ public String getTableLocation() {
3939

4040
@Override
4141
public boolean isSupported() {
42-
return false;
42+
return true;
4343
}
4444
}

‎fe/fe-core/src/main/java/com/starrocks/planner/LanceScanNode.java‎

Lines changed: 227 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,227 @@
1+
// Copyright 2021-present StarRocks, Inc. All rights reserved.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package com.starrocks.planner;
16+
17+
import com.google.common.annotations.VisibleForTesting;
18+
import com.google.common.base.MoreObjects;
19+
import com.google.common.base.Preconditions;
20+
import com.starrocks.catalog.LanceTable;
21+
import com.starrocks.connector.CatalogConnector;
22+
import com.starrocks.credential.CloudConfiguration;
23+
import com.starrocks.qe.ConnectContext;
24+
import com.starrocks.server.GlobalStateMgr;
25+
import com.starrocks.server.RunMode;
26+
import com.starrocks.server.WarehouseManager;
27+
import com.starrocks.sql.optimizer.operator.scalar.ScalarOperator;
28+
import com.starrocks.sql.plan.HDFSScanNodePredicates;
29+
import com.starrocks.system.ComputeNode;
30+
import com.starrocks.system.SystemInfoService;
31+
import com.starrocks.thrift.TConnectorScanNode;
32+
import com.starrocks.thrift.TExplainLevel;
33+
import com.starrocks.thrift.THdfsFileFormat;
34+
import com.starrocks.thrift.THdfsScanNode;
35+
import com.starrocks.thrift.THdfsScanRange;
36+
import com.starrocks.thrift.TPlanNode;
37+
import com.starrocks.thrift.TPlanNodeType;
38+
import com.starrocks.thrift.TScanRange;
39+
import com.starrocks.thrift.TScanRangeLocation;
40+
import com.starrocks.thrift.TScanRangeLocations;
41+
import com.starrocks.type.Type;
42+
import com.starrocks.warehouse.cngroup.ComputeResource;
43+
import org.apache.logging.log4j.LogManager;
44+
import org.apache.logging.log4j.Logger;
45+
46+
import java.util.ArrayList;
47+
import java.util.List;
48+
import java.util.stream.Collectors;
49+
50+
import static com.starrocks.thrift.TExplainLevel.VERBOSE;
51+
52+
public class LanceScanNode extends ScanNode {
53+
private static final Logger LOG = LogManager.getLogger(LanceScanNode.class);
54+
private final LanceTable lanceTable;
55+
private final HDFSScanNodePredicates scanNodePredicates = new HDFSScanNodePredicates();
56+
private final List<TScanRangeLocations> scanRangeLocationsList = new ArrayList<>();
57+
private CloudConfiguration cloudConfiguration = null;
58+
59+
public LanceScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeName) {
60+
super(id, desc, planNodeName);
61+
this.lanceTable = (LanceTable) desc.getTable();
62+
setupCloudCredential();
63+
}
64+
65+
public HDFSScanNodePredicates getScanNodePredicates() {
66+
return scanNodePredicates;
67+
}
68+
69+
private void setupCloudCredential() {
70+
// Lance tables will use properties/credentials defined in catalog
71+
// For Phase 1 & 2 mock catalog, we can safely allow empty config or resolve standard connectors if registered
72+
}
73+
74+
@Override
75+
protected String debugString() {
76+
MoreObjects.ToStringHelper helper = MoreObjects.toStringHelper(this);
77+
helper.addValue(super.debugString());
78+
helper.addValue("lanceTable=" + lanceTable.getName());
79+
return helper.toString();
80+
}
81+
82+
@Override
83+
public List<TScanRangeLocations> getScanRangeLocations(long maxScanRangeLength) {
84+
return scanRangeLocationsList;
85+
}
86+
87+
public void setupScanRangeLocations(TupleDescriptor tupleDescriptor, ScalarOperator predicate) {
88+
List<Long> nodeIds = getAllAvailableBackendOrComputeIds();
89+
if (nodeIds.isEmpty()) {
90+
return;
91+
}
92+
93+
TScanRangeLocations scanRangeLocations = new TScanRangeLocations();
94+
95+
THdfsScanRange hdfsScanRange = new THdfsScanRange();
96+
hdfsScanRange.setUse_lance_jni_reader(true);
97+
hdfsScanRange.setLance_dataset_uri(lanceTable.getUri());
98+
hdfsScanRange.setLance_split_info("{\"fragment_ids\": [0]}");
99+
hdfsScanRange.setFile_length(0);
100+
hdfsScanRange.setLength(0);
101+
hdfsScanRange.setFile_format(THdfsFileFormat.UNKNOWN);
102+
103+
TScanRange scanRange = new TScanRange();
104+
scanRange.setHdfs_scan_range(hdfsScanRange);
105+
scanRangeLocations.setScan_range(scanRange);
106+
107+
TScanRangeLocation scanRangeLocation = new TScanRangeLocation();
108+
scanRangeLocation.setBackend_id(nodeIds.get(0));
109+
scanRangeLocations.addToLocations(scanRangeLocation);
110+
scanRangeLocationsList.add(scanRangeLocations);
111+
}
112+
113+
@VisibleForTesting
114+
public List<Long> getAllAvailableBackendOrComputeIds() {
115+
List<Long> allNodes = new ArrayList<>();
116+
SystemInfoService systemInfoService = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();
117+
if (RunMode.isSharedDataMode()) {
118+
ComputeResource computeResource = WarehouseManager.DEFAULT_RESOURCE;
119+
if (ConnectContext.get() != null) {
120+
computeResource = ConnectContext.get().getCurrentComputeResource();
121+
}
122+
final WarehouseManager warehouseManager = GlobalStateMgr.getCurrentState().getWarehouseMgr();
123+
allNodes = warehouseManager.getAliveComputeNodes(computeResource)
124+
.stream()
125+
.map(ComputeNode::getId)
126+
.collect(Collectors.toList());
127+
} else {
128+
allNodes = systemInfoService.getAvailableBackendIds();
129+
if (allNodes == null) {
130+
allNodes = new ArrayList<>();
131+
} else {
132+
allNodes = new ArrayList<>(allNodes);
133+
}
134+
List<Long> computeNodeIds = systemInfoService.getAvailableComputeNodeIds();
135+
if (computeNodeIds != null) {
136+
allNodes.addAll(computeNodeIds);
137+
}
138+
}
139+
if (allNodes.isEmpty()) {
140+
allNodes.add(10001L); // Fallback for unit testing where cluster state is empty
141+
}
142+
return allNodes;
143+
}
144+
145+
@Override
146+
protected String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
147+
StringBuilder output = new StringBuilder();
148+
149+
output.append(prefix).append("TABLE: ").append(lanceTable.getName()).append("\n");
150+
151+
if (null != sortColumn) {
152+
output.append(prefix).append("SORT COLUMN: ").append(sortColumn).append("\n");
153+
}
154+
if (!conjuncts.isEmpty()) {
155+
output.append(prefix).append("PREDICATES: ").append(
156+
explainExpr(conjuncts)).append("\n");
157+
}
158+
if (!scanNodePredicates.getPartitionConjuncts().isEmpty()) {
159+
output.append(prefix).append("PARTITION PREDICATES: ").append(
160+
explainExpr(scanNodePredicates.getPartitionConjuncts())).append("\n");
161+
}
162+
if (!scanNodePredicates.getNonPartitionConjuncts().isEmpty()) {
163+
output.append(prefix).append("NON-PARTITION PREDICATES: ").append(
164+
explainExpr(scanNodePredicates.getNonPartitionConjuncts())).append("\n");
165+
}
166+
if (!scanNodePredicates.getNoEvalPartitionConjuncts().isEmpty()) {
167+
output.append(prefix).append("NO EVAL-PARTITION PREDICATES: ").append(
168+
explainExpr(scanNodePredicates.getNoEvalPartitionConjuncts())).append("\n");
169+
}
170+
if (!scanNodePredicates.getMinMaxConjuncts().isEmpty()) {
171+
output.append(prefix).append("MIN/MAX PREDICATES: ").append(
172+
explainExpr(scanNodePredicates.getMinMaxConjuncts())).append("\n");
173+
}
174+
175+
if (detailLevel != VERBOSE) {
176+
output.append(prefix).append(String.format("cardinality=%s", cardinality));
177+
output.append("\n");
178+
}
179+
180+
output.append("\n");
181+
output.append(prefix).append(String.format("avgRowSize=%s\n", avgRowSize));
182+
183+
if (detailLevel == TExplainLevel.VERBOSE) {
184+
for (SlotDescriptor slotDescriptor : desc.getSlots()) {
185+
Type type = slotDescriptor.getOriginType();
186+
if (type.isComplexType()) {
187+
output.append(prefix)
188+
.append(String.format("Pruned type: %d <-> [%s]\n", slotDescriptor.getId().asInt(), type));
189+
}
190+
}
191+
}
192+
193+
return output.toString();
194+
}
195+
196+
@Override
197+
protected void toThrift(TPlanNode msg) {
198+
msg.node_type = TPlanNodeType.HDFS_SCAN_NODE;
199+
THdfsScanNode tHdfsScanNode = new THdfsScanNode();
200+
tHdfsScanNode.setTuple_id(desc.getId().asInt());
201+
msg.hdfs_scan_node = tHdfsScanNode;
202+
203+
TConnectorScanNode connectorScanNode = new TConnectorScanNode();
204+
connectorScanNode.setConnector_name("lance");
205+
msg.connector_scan_node = connectorScanNode;
206+
207+
String sqlPredicates = getExplainString(conjuncts);
208+
msg.hdfs_scan_node.setSql_predicates(sqlPredicates);
209+
210+
if (lanceTable != null) {
211+
msg.hdfs_scan_node.setTable_name(lanceTable.getName());
212+
}
213+
214+
HdfsScanNode.setScanOptimizeOptionToThrift(tHdfsScanNode, this);
215+
HdfsScanNode.setCloudConfigurationToThrift(tHdfsScanNode, cloudConfiguration);
216+
HdfsScanNode.setMinMaxConjunctsToThrift(tHdfsScanNode, this, this.getScanNodePredicates());
217+
HdfsScanNode.setNonEvalPartitionConjunctsToThrift(tHdfsScanNode, this, this.getScanNodePredicates());
218+
HdfsScanNode.setNonPartitionConjunctsToThrift(msg, this, this.getScanNodePredicates());
219+
220+
setConnectorCatalogType(msg);
221+
}
222+
223+
@Override
224+
public boolean canUseRuntimeAdaptiveDop() {
225+
return true;
226+
}
227+
}

‎fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@
8888
import com.starrocks.planner.JDBCScanNode;
8989
import com.starrocks.planner.JoinNode;
9090
import com.starrocks.planner.KuduScanNode;
91+
import com.starrocks.planner.LanceScanNode;
9192
import com.starrocks.planner.LookUpNode;
9293
import com.starrocks.planner.MergeJoinNode;
9394
import com.starrocks.planner.MetaScanNode;
@@ -193,6 +194,7 @@
193194
import com.starrocks.sql.optimizer.operator.physical.PhysicalJDBCScanOperator;
194195
import com.starrocks.sql.optimizer.operator.physical.PhysicalJoinOperator;
195196
import com.starrocks.sql.optimizer.operator.physical.PhysicalKuduScanOperator;
197+
import com.starrocks.sql.optimizer.operator.physical.PhysicalLanceScanOperator;
196198
import com.starrocks.sql.optimizer.operator.physical.PhysicalLimitOperator;
197199
import com.starrocks.sql.optimizer.operator.physical.PhysicalLookUpOperator;
198200
import com.starrocks.sql.optimizer.operator.physical.PhysicalMergeJoinOperator;
@@ -1720,6 +1722,49 @@ public PlanFragment visitPhysicalKuduScan(OptExpression optExpression, ExecPlan
17201722
return fragment;
17211723
}
17221724

1725+
public PlanFragment visitPhysicalLanceScan(OptExpression optExpression, ExecPlan context) {
1726+
PhysicalLanceScanOperator node = (PhysicalLanceScanOperator) optExpression.getOp();
1727+
1728+
Table referenceTable = node.getTable();
1729+
context.getDescTbl().addReferencedTable(referenceTable);
1730+
TupleDescriptor tupleDescriptor = context.getDescTbl().createTupleDescriptor();
1731+
tupleDescriptor.setTable(referenceTable);
1732+
1733+
// set slot
1734+
prepareContextSlots(node, context, tupleDescriptor);
1735+
1736+
LanceScanNode lanceScanNode =
1737+
new LanceScanNode(context.getNextNodeId(), tupleDescriptor, "LanceScanNode");
1738+
lanceScanNode.setScanOptimizeOption(node.getScanOptimizeOption());
1739+
try {
1740+
// set predicate
1741+
ScalarOperatorToExpr.FormatterContext formatterContext =
1742+
new ScalarOperatorToExpr.FormatterContext(context.getColRefToExpr());
1743+
List<ScalarOperator> predicates = Utils.extractConjuncts(node.getPredicate());
1744+
for (ScalarOperator predicate : predicates) {
1745+
lanceScanNode.getConjuncts()
1746+
.add(ScalarOperatorToExpr.buildExecExpression(predicate, formatterContext));
1747+
}
1748+
lanceScanNode.setupScanRangeLocations(tupleDescriptor, node.getPredicate());
1749+
HDFSScanNodePredicates scanNodePredicates = lanceScanNode.getScanNodePredicates();
1750+
prepareMinMaxExpr(scanNodePredicates, node.getScanOperatorPredicates(), context, referenceTable);
1751+
prepareCommonExpr(scanNodePredicates, node.getScanOperatorPredicates(), context);
1752+
} catch (Exception e) {
1753+
LOG.warn("Lance scan node get scan range locations failed : ", e);
1754+
throw new StarRocksPlannerException(e.getMessage(), INTERNAL_ERROR);
1755+
}
1756+
1757+
lanceScanNode.setLimit(node.getLimit());
1758+
1759+
tupleDescriptor.computeMemLayout();
1760+
registerScanNode(node, lanceScanNode, context);
1761+
1762+
PlanFragment fragment =
1763+
new PlanFragment(context.getNextFragmentId(), lanceScanNode, DataPartition.RANDOM);
1764+
context.getFragments().add(fragment);
1765+
return fragment;
1766+
}
1767+
17231768
@Override
17241769
public PlanFragment visitPhysicalIcebergScan(OptExpression optExpression, ExecPlan context) {
17251770
return buildIcebergScanNode(optExpression, context);

‎fe/fe-core/src/test/java/com/starrocks/common/util/ProfilingExecPlanTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ public void testNormalizeName() throws Exception {
7575
"STREAM_LOAD_SCAN", "ANALYTIC_EVAL", "ICEBERG_SCAN", "AGGREGATION", "FILE_SCAN", "EXCHANGE",
7676
"META_SCAN", "OLAP_SCAN", "ODPS_SCAN", "ICEBERG_METADATA_SCAN", "KUDU_SCAN",
7777
"CAPTURE_VERSION", "ICEBERG_EQUALITY_DELETE_SCAN", "RAW_VALUES", "FETCH", "LOOK_UP",
78-
"BENCHMARK_SCAN", "ICEBERG_METADATA_DELETE", "CACHE_STATS_SCAN");
78+
"BENCHMARK_SCAN", "ICEBERG_METADATA_DELETE", "CACHE_STATS_SCAN", "LANCE_SCAN");
7979

8080
Method method = ProfilingExecPlan.class.getDeclaredMethod("normalizeNodeName", Class.class);
8181
method.setAccessible(true);

‎fe/fe-core/src/test/java/com/starrocks/connector/lance/LanceMetadataTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public void testMetadataDiscoveryBootstrap() {
102102
LanceTable lanceDiscovered = (LanceTable) discovered;
103103
Assertions.assertEquals("s3://bucket/users", lanceDiscovered.getUri());
104104
Assertions.assertEquals("s3://bucket/users", lanceDiscovered.getTableLocation());
105-
Assertions.assertFalse(lanceDiscovered.isSupported()); // verified unsupported in planning
105+
Assertions.assertTrue(lanceDiscovered.isSupported());
106106

107107
Table events = metadata.getTable(null, "vectors_db", "events");
108108
Assertions.assertNotNull(events);

0 commit comments

Comments
 (0)