Skip to content

[bugfix](topn) cherry pick key topn fixes for branch-1.2-lts #20406

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morningman merged 5 commits into from
Jun 6, 2023
Merged

[bugfix](topn) cherry pick key topn fixes for branch-1.2-lts #20406

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
253a306
[Improvement](topn) add limit threashold session variable and fuzzy f…
xiaokang Feb 10, 2023
8bb20aa
partital pick for key topn only:
xiaokang Jun 4, 2023
2b56412
[bugfix](topn) fix topn read_orderby_key_columns nullptr (#16896)
xiaokang Jun 4, 2023
da412f5
disable key topn opt only for DUP_KEYS and UNIQUE_KEYS with MOW
xiaokang Jun 4, 2023
6bcd3a5
add testcase for key topn opt
xiaokang Jun 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 2 additions & 5 deletions be/src/olap/reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,12 +402,9 @@ Status TabletReader::_init_keys_param(const ReaderParams& read_params) {
}

Status TabletReader::_init_orderby_keys_param(const ReaderParams& read_params) {
if (read_params.start_key.empty()) {
return Status::OK();
}

// UNIQUE_KEYS will compare all keys as before
if (_tablet_schema->keys_type() == DUP_KEYS) {
if (_tablet_schema->keys_type() == DUP_KEYS || (_tablet_schema->keys_type() == UNIQUE_KEYS &&
_tablet->enable_unique_key_merge_on_write())) {
// find index in vector _return_columns
// for the read_orderby_key_num_prefix_columns orderby keys
for (uint32_t i = 0; i < read_params.read_orderby_key_num_prefix_columns; i++) {
Expand Down
6 changes: 6 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1985,4 +1985,10 @@ public void initSchemaColumnUniqueId() {
public Set<Long> getPartitionKeys() {
return idToPartition.keySet();
}

public boolean isDupKeysOrMergeOnWrite() {
return getKeysType() == KeysType.DUP_KEYS
|| (getKeysType() == KeysType.UNIQUE_KEYS
&& getEnableUniqueKeyMergeOnWrite());
}
}
25 changes: 24 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -928,6 +928,16 @@ private void computeTabletInfo() throws UserException {
* Check Parent sort node can push down to child olap scan.
*/
public boolean checkPushSort(SortNode sortNode) {
if (!olapTable.isDupKeysOrMergeOnWrite()) {
return false;
}

// Ensure limit is less then threshold
if (sortNode.getLimit() <= 0
|| sortNode.getLimit() > ConnectContext.get().getSessionVariable().topnOptLimitThreshold) {
return false;
}

// Ensure all isAscOrder is same, ande length != 0.
// Can't be zorder.
if (sortNode.getSortInfo().getIsAscOrder().stream().distinct().count() != 1
Expand All @@ -942,6 +952,8 @@ public boolean checkPushSort(SortNode sortNode) {
// sort key: (a) (a,b) (a,b,c) (a,b,c,d) is ok
// (a,c) (a,c,d), (a,c,b) (a,c,f) (a,b,c,d,e)is NOT ok
List<Expr> sortExprs = sortNode.getSortInfo().getMaterializedOrderingExprs();
List<Boolean> nullsFirsts = sortNode.getSortInfo().getNullsFirst();
List<Boolean> isAscOrders = sortNode.getSortInfo().getIsAscOrder();
if (sortExprs.size() > olapTable.getDataSortInfo().getColNum()) {
return false;
}
Expand All @@ -950,7 +962,18 @@ public boolean checkPushSort(SortNode sortNode) {
Column tableKey = olapTable.getFullSchema().get(i);
// sort slot.
Expr sortExpr = sortExprs.get(i);
if (!(sortExpr instanceof SlotRef) || !tableKey.equals(((SlotRef) sortExpr).getColumn())) {
if (sortExpr instanceof SlotRef) {
SlotRef slotRef = (SlotRef) sortExpr;
if (tableKey.equals(slotRef.getColumn())) {
// ORDER BY DESC NULLS FIRST can not be optimized to only read file tail,
// since NULLS is at file head but data is at tail
if (tableKey.isAllowNull() && nullsFirsts.get(i) && !isAscOrders.get(i)) {
return false;
}
} else {
return false;
}
} else {
return false;
}
}
Expand Down
7 changes: 7 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,8 @@ public class SessionVariable implements Serializable, Writable {

public static final String GROUP_CONCAT_MAX_LEN = "group_concat_max_len";

public static final String TOPN_OPT_LIMIT_THRESHOLD = "topn_opt_limit_threshold";

public static final String GROUP_BY_AND_HAVING_USE_ALIAS_FIRST = "group_by_and_having_use_alias_first";

// fix replica to query. If num = 1, query the smallest replica, if 2 is the second smallest replica.
Expand Down Expand Up @@ -671,6 +673,9 @@ public class SessionVariable implements Serializable, Writable {
@VariableMgr.VarAttr(name = GROUP_CONCAT_MAX_LEN)
public long groupConcatMaxLen = 2147483646;

@VariableMgr.VarAttr(name = TOPN_OPT_LIMIT_THRESHOLD)
public long topnOptLimitThreshold = 1024;

// Default value is false, which means the group by and having clause
// should first use column name not alias. According to mysql.
@VariableMgr.VarAttr(name = GROUP_BY_AND_HAVING_USE_ALIAS_FIRST)
Expand All @@ -697,6 +702,8 @@ public void initFuzzyModeVariables() {
} else {
this.rewriteOrToInPredicateThreshold = 2;
}
// set random 1, 10, 100, 1000, 10000
this.topnOptLimitThreshold = (int) Math.pow(10, random.nextInt(5));
}

public String printFuzzyVariables() {
Expand Down
76 changes: 60 additions & 16 deletions fe/fe-core/src/test/java/org/apache/doris/planner/PlannerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -508,17 +508,49 @@ public void testUpdateUnique() throws Exception {

@Test
public void testPushSortToOlapScan() throws Exception {
// Push sort successfully
String sql1 = "explain select k1 from db1.tbl1 order by k1, k2";
// Push sort fail without limit
String sql1 = "explain select k1 from db1.tbl3 order by k1, k2";
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
String plan1 = planner1.getExplainString(new ExplainOptions(false, false));
Assertions.assertFalse(plan1.contains("SORT INFO:\n `k1`\n `k2`"));
Assertions.assertFalse(plan1.contains("SORT LIMIT:"));

// Push sort fail limit > topnOptLimitThreshold
sql1 = "explain select k1 from db1.tbl3 order by k1, k2 limit "
+ (connectContext.getSessionVariable().topnOptLimitThreshold + 1);
stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
planner1 = stmtExecutor1.planner();
plan1 = planner1.getExplainString(new ExplainOptions(false, false));
Assertions.assertFalse(plan1.contains("SORT INFO:\n `k1`\n `k2`"));
Assertions.assertFalse(plan1.contains("SORT LIMIT:"));

// Push sort success limit = topnOptLimitThreshold
sql1 = "explain select k1 from db1.tbl3 order by k1, k2 limit "
+ (connectContext.getSessionVariable().topnOptLimitThreshold);
stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
planner1 = stmtExecutor1.planner();
plan1 = planner1.getExplainString(new ExplainOptions(false, false));
Assertions.assertTrue(plan1.contains("SORT INFO:\n `k1`\n `k2`"));
Assertions.assertTrue(plan1.contains("SORT LIMIT:"));

// Push sort success limit < topnOptLimitThreshold
if (connectContext.getSessionVariable().topnOptLimitThreshold > 1) {
sql1 = "explain select k1 from db1.tbl3 order by k1, k2 limit "
+ (connectContext.getSessionVariable().topnOptLimitThreshold - 1);
stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
planner1 = stmtExecutor1.planner();
plan1 = planner1.getExplainString(new ExplainOptions(false, false));
Assertions.assertTrue(plan1.contains("SORT INFO:\n `k1`\n `k2`"));
Assertions.assertTrue(plan1.contains("SORT LIMIT:"));
}

// Push sort failed
String sql2 = "explain select k1, k2, k3 from db1.tbl1 order by k1, k3, k2";
String sql2 = "explain select k1, k2, k3 from db1.tbl3 order by k1, k3, k2";
StmtExecutor stmtExecutor2 = new StmtExecutor(connectContext, sql2);
stmtExecutor2.execute();
Planner planner2 = stmtExecutor2.planner();
Expand All @@ -529,9 +561,10 @@ public void testPushSortToOlapScan() throws Exception {

@Test
public void testEliminatingSortNode() throws Exception {
// fail case 1
// success case 1
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -542,7 +575,8 @@ public void testEliminatingSortNode() throws Exception {

// fail case 2
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k3 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k3 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -553,7 +587,8 @@ public void testEliminatingSortNode() throws Exception {

// fail case 3
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 != 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 != 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -564,7 +599,8 @@ public void testEliminatingSortNode() throws Exception {

// fail case 4
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 or k2 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 or k2 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -575,7 +611,8 @@ public void testEliminatingSortNode() throws Exception {

// fail case 5
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 = 2 or k3 = 3 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 = 2 or k3 = 3 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -587,7 +624,8 @@ public void testEliminatingSortNode() throws Exception {
// fail case 6
// TODO, support: in (select 1)
{
String sql1 = "explain select k1 from db1.tbl1 where k1 in (select 1) and k2 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 in (select 1) and k2 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -597,7 +635,8 @@ public void testEliminatingSortNode() throws Exception {

// fail case 7
{
String sql1 = "explain select k1 from db1.tbl1 where k1 not in (1) and k2 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 not in (1) and k2 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -607,7 +646,8 @@ public void testEliminatingSortNode() throws Exception {

// success case 1
{
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 = 1 and k2 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -618,7 +658,8 @@ public void testEliminatingSortNode() throws Exception {

// success case 2
{
String sql1 = "explain select k1 from db1.tbl1 where k3 = 3 and k2 = 2 and k1 = 1 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k3 = 3 and k2 = 2 and k1 = 1 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -629,7 +670,8 @@ public void testEliminatingSortNode() throws Exception {

// success case 3
{
String sql1 = "explain select k1 from db1.tbl1 where k1 in (1) and k2 in (2) and k2 !=2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 in (1) and k2 in (2) and k2 !=2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -640,7 +682,8 @@ public void testEliminatingSortNode() throws Exception {

// success case 4
{
String sql1 = "explain select k1 from db1.tbl1 where k1 in (concat('1','2')) and k2 = 2 order by k1, k2";
String sql1 = "explain select k1 from db1.tbl1 where k1 in (concat('1','2')) and k2 = 2 order by k1, k2 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand All @@ -652,7 +695,8 @@ public void testEliminatingSortNode() throws Exception {
// success case 5
{
String sql1 = "explain select tbl1.k1 from db1.tbl1 join db1.tbl2 on tbl1.k1 = tbl2.k1"
+ " where tbl1.k1 = 1 and tbl2.k1 = 2 and tbl1.k2 = 3 order by tbl1.k1, tbl2.k1";
+ " where tbl1.k1 = 1 and tbl2.k1 = 2 and tbl1.k2 = 3 order by tbl1.k1, tbl2.k1 limit "
+ connectContext.getSessionVariable().topnOptLimitThreshold;
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
Expand Down
Loading