Skip to content

Commit 2467570

Browse files
natalya-aksmannatalya-aksman
committed
Do not sort batches for Batch sorted merge over unordered chunks if query sort matches compressed sort
1 parent f46f9e0 commit 2467570

8 files changed

Lines changed: 1168 additions & 254 deletions

File tree

‎tsl/src/nodes/columnar_scan/columnar_scan.c‎

Lines changed: 186 additions & 107 deletions
Original file line numberDiff line numberDiff line change
@@ -241,7 +241,7 @@ build_compressed_scan_pathkeys(const SortInfo *sort_info, PlannerInfo *root, Lis
241241
* If pathkeys contains non-segmentby columns the rest of the ordering
242242
* requirements will be satisfied by ordering by sequence_num.
243243
*/
244-
if (sort_info->needs_sequence_num)
244+
if (sort_info->needs_sequence_num || sort_info->use_batch_sorted_merge)
245245
{
246246
/* TODO: split up legacy sequence number path and non-sequence number path into dedicated
247247
* functions. */
@@ -317,60 +317,10 @@ build_compressed_scan_pathkeys(const SortInfo *sort_info, PlannerInfo *root, Lis
317317
column_name = get_attname(info->chunk_rte->relid, var->varattno, false);
318318
int16 orderby_index = ts_array_position(info->settings->fd.orderby, column_name);
319319
Assert(orderby_index != 0);
320-
AttrNumber leading_attno;
321-
AttrNumber trailing_attno;
322-
orderby_sparse_metadata_attnos(info->settings,
323-
info->compressed_rte->relid,
324-
orderby_index,
325-
&leading_attno,
326-
&trailing_attno);
320+
327321
bool orderby_desc =
328322
ts_array_get_element_bool(info->settings->fd.orderby_desc, orderby_index);
329323

330-
/*
331-
* Compressed chunk indexes based on firstlast sparse indexes can have two
332-
* orderings. New chunks index them as (first, last); chunks compressed before that
333-
* change index a DESC column as (last, first). Only DESC columns can differ, so for
334-
* those we check the chunk's index and follow whichever order it has.
335-
*/
336-
if (orderby_desc &&
337-
orderby_sparse_kind(info->settings, orderby_index) == ORDERBY_SPARSE_FIRSTLAST)
338-
{
339-
orderby_firstlast_metadata_attnos(info->settings,
340-
info->compressed_rte->relid,
341-
orderby_index,
342-
&leading_attno,
343-
&trailing_attno);
344-
345-
ListCell *index_lc;
346-
foreach (index_lc, info->compressed_rel->indexlist)
347-
{
348-
IndexOptInfo *index = lfirst(index_lc);
349-
int leading_pos = -1;
350-
int trailing_pos = -1;
351-
for (int k = 0; k < index->nkeycolumns; k++)
352-
{
353-
if (index->indexkeys[k] == leading_attno)
354-
{
355-
leading_pos = k;
356-
}
357-
else if (index->indexkeys[k] == trailing_attno)
358-
{
359-
trailing_pos = k;
360-
}
361-
}
362-
if (leading_pos >= 0 && trailing_pos >= 0)
363-
{
364-
if (trailing_pos < leading_pos)
365-
{
366-
AttrNumber tmp = leading_attno;
367-
leading_attno = trailing_attno;
368-
trailing_attno = tmp;
369-
}
370-
break;
371-
}
372-
}
373-
}
374324
bool orderby_nullsfirst =
375325
ts_array_get_element_bool(info->settings->fd.orderby_nullsfirst, orderby_index);
376326

@@ -388,43 +338,151 @@ build_compressed_scan_pathkeys(const SortInfo *sort_info, PlannerInfo *root, Lis
388338
nulls_first = orderby_nullsfirst;
389339
}
390340

391-
Var *metadata_var = makeVar(info->compressed_rel->relid,
392-
leading_attno,
393-
var->vartype,
394-
var->vartypmod,
395-
var->varcollid,
396-
var->varlevelsup);
397-
Expr *leading_expr =
398-
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
399-
EquivalenceClass *leading_ec =
400-
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
401-
PathKey *leading_pk = make_canonical_pathkey(root,
402-
leading_ec,
403-
pk->pk_opfamily,
404-
strategy,
405-
nulls_first);
406-
required_compressed_pathkeys = lappend(required_compressed_pathkeys, leading_pk);
407-
408-
metadata_var = makeVar(info->compressed_rel->relid,
409-
trailing_attno,
410-
var->vartype,
411-
var->vartypmod,
412-
var->varcollid,
413-
var->varlevelsup);
414-
Expr *trailing_expr =
415-
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
416-
EquivalenceClass *trailing_ec =
417-
append_ec_for_metadata_col(root, info, trailing_expr, pk, opcintype);
418-
PathKey *trailing_pk = make_canonical_pathkey(root,
419-
trailing_ec,
420-
pk->pk_opfamily,
421-
strategy,
422-
nulls_first);
423-
424-
required_compressed_pathkeys = lappend(required_compressed_pathkeys, trailing_pk);
341+
/* For Batch sorted merge we need to sort on specially chosen leading attribute for
342+
* each pathkey */
343+
if (sort_info->use_batch_sorted_merge)
344+
{
345+
Oid sortop =
346+
get_opfamily_member(pk->pk_opfamily, opcintype, opcintype, pk->pk_cmptype);
347+
Oid opfamily, optype;
348+
CompareType bsm_strategy;
349+
if (!get_ordering_op_properties(sortop, &opfamily, &optype, &bsm_strategy))
350+
{
351+
elog(ERROR, "operator %u is not a valid ordering operator", sortop);
352+
}
353+
Assert(bsm_strategy == BTLessStrategyNumber ||
354+
bsm_strategy == BTGreaterStrategyNumber);
355+
char *leading_name;
356+
char *trailing_name;
357+
orderby_sparse_metadata_names(info->settings,
358+
orderby_index,
359+
&leading_name,
360+
&trailing_name);
361+
char *meta_col_name =
362+
strategy == BTLessStrategyNumber ? leading_name : trailing_name;
363+
364+
AttrNumber attr_position =
365+
get_attnum(info->compressed_rte->relid, meta_col_name);
366+
367+
if (attr_position == InvalidAttrNumber)
368+
{
369+
elog(ERROR, "couldn't find metadata column \"%s\"", meta_col_name);
370+
}
371+
Var *metadata_var = makeVar(info->compressed_rel->relid,
372+
attr_position,
373+
var->vartype,
374+
var->vartypmod,
375+
var->varcollid,
376+
var->varlevelsup);
377+
Expr *leading_expr =
378+
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
379+
EquivalenceClass *leading_ec =
380+
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
381+
PathKey *leading_pk = make_canonical_pathkey(root,
382+
leading_ec,
383+
pk->pk_opfamily,
384+
strategy,
385+
nulls_first);
386+
required_compressed_pathkeys =
387+
lappend(required_compressed_pathkeys, leading_pk);
388+
}
389+
/* Need to sort on compressed pathkeys matching compressed indexscan order */
390+
else
391+
{
392+
AttrNumber leading_attno;
393+
AttrNumber trailing_attno;
394+
orderby_sparse_metadata_attnos(info->settings,
395+
info->compressed_rte->relid,
396+
orderby_index,
397+
&leading_attno,
398+
&trailing_attno);
399+
/*
400+
* Compressed chunk indexes based on firstlast sparse indexes can have two
401+
* orderings. New chunks index them as (first, last); chunks compressed before
402+
* that change index a DESC column as (last, first). Only DESC columns can
403+
* differ, so for those we check the chunk's index and follow whichever order it
404+
* has.
405+
*/
406+
if (orderby_desc && orderby_sparse_kind(info->settings, orderby_index) ==
407+
ORDERBY_SPARSE_FIRSTLAST)
408+
{
409+
orderby_firstlast_metadata_attnos(info->settings,
410+
info->compressed_rte->relid,
411+
orderby_index,
412+
&leading_attno,
413+
&trailing_attno);
414+
415+
ListCell *index_lc;
416+
foreach (index_lc, info->compressed_rel->indexlist)
417+
{
418+
IndexOptInfo *index = lfirst(index_lc);
419+
int leading_pos = -1;
420+
int trailing_pos = -1;
421+
for (int k = 0; k < index->nkeycolumns; k++)
422+
{
423+
if (index->indexkeys[k] == leading_attno)
424+
{
425+
leading_pos = k;
426+
}
427+
else if (index->indexkeys[k] == trailing_attno)
428+
{
429+
trailing_pos = k;
430+
}
431+
}
432+
if (leading_pos >= 0 && trailing_pos >= 0)
433+
{
434+
if (trailing_pos < leading_pos)
435+
{
436+
AttrNumber tmp = leading_attno;
437+
leading_attno = trailing_attno;
438+
trailing_attno = tmp;
439+
}
440+
break;
441+
}
442+
}
443+
}
444+
Var *metadata_var;
445+
metadata_var = makeVar(info->compressed_rel->relid,
446+
leading_attno,
447+
var->vartype,
448+
var->vartypmod,
449+
var->varcollid,
450+
var->varlevelsup);
451+
Expr *leading_expr =
452+
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
453+
EquivalenceClass *leading_ec =
454+
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
455+
PathKey *leading_pk = make_canonical_pathkey(root,
456+
leading_ec,
457+
pk->pk_opfamily,
458+
strategy,
459+
nulls_first);
460+
required_compressed_pathkeys =
461+
lappend(required_compressed_pathkeys, leading_pk);
462+
463+
metadata_var = makeVar(info->compressed_rel->relid,
464+
trailing_attno,
465+
var->vartype,
466+
var->vartypmod,
467+
var->varcollid,
468+
var->varlevelsup);
469+
Expr *trailing_expr =
470+
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
471+
EquivalenceClass *trailing_ec =
472+
append_ec_for_metadata_col(root, info, trailing_expr, pk, opcintype);
473+
PathKey *trailing_pk = make_canonical_pathkey(root,
474+
trailing_ec,
475+
pk->pk_opfamily,
476+
strategy,
477+
nulls_first);
478+
479+
required_compressed_pathkeys =
480+
lappend(required_compressed_pathkeys, trailing_pk);
481+
}
425482
}
426483
}
427484
}
485+
428486
return required_compressed_pathkeys;
429487
}
430488

@@ -978,27 +1036,45 @@ cost_batch_sorted_merge(PlannerInfo *root, const CompressionInfo *compression_in
9781036
{
9791037
Path sort_path; /* dummy for result of cost_sort */
9801038

981-
/*
982-
* Don't disable the compressed batch sorted merge plan with the enable_sort
983-
* GUC. We have a separate GUC for it, and this way you can try to force the
984-
* batch sorted merge plan by disabling sort.
985-
*/
986-
const bool old_enable_sort = enable_sort;
987-
enable_sort = true;
988-
cost_sort(&sort_path,
989-
root,
990-
dcpath->required_compressed_pathkeys,
1039+
/* We are utilizing compressed sort for batch sorted merge: do not need extra sort */
1040+
if (dcpath->required_compressed_pathkeys &&
1041+
pathkeys_contained_in(dcpath->required_compressed_pathkeys, compressed_path->pathkeys))
1042+
{
1043+
sort_path.rows = compressed_path->rows;
1044+
sort_path.startup_cost = compressed_path->startup_cost;
1045+
sort_path.total_cost = compressed_path->total_cost;
9911046
#if PG18_GE
992-
compressed_path->disabled_nodes,
1047+
/* PG18 changes the way we handle disabled nodes so we
1048+
* need to take those into account as well.
1049+
*
1050+
* https://github.com/postgres/postgres/commit/e2225346
1051+
*/
1052+
sort_path.disabled_nodes = compressed_path->disabled_nodes;
9931053
#endif
994-
compressed_path->total_cost,
995-
compressed_path->rows,
996-
compressed_path->pathtarget->width,
997-
0.0,
998-
work_mem,
999-
-1);
1000-
enable_sort = old_enable_sort;
1001-
1054+
}
1055+
else
1056+
{
1057+
/*
1058+
* Don't disable the compressed batch sorted merge plan with the enable_sort
1059+
* GUC. We have a separate GUC for it, and this way you can try to force the
1060+
* batch sorted merge plan by disabling sort.
1061+
*/
1062+
const bool old_enable_sort = enable_sort;
1063+
enable_sort = true;
1064+
cost_sort(&sort_path,
1065+
root,
1066+
dcpath->required_compressed_pathkeys,
1067+
#if PG18_GE
1068+
compressed_path->disabled_nodes,
1069+
#endif
1070+
compressed_path->total_cost,
1071+
compressed_path->rows,
1072+
compressed_path->pathtarget->width,
1073+
0.0,
1074+
work_mem,
1075+
-1);
1076+
enable_sort = old_enable_sort;
1077+
}
10021078
/*
10031079
* In compressed batch sorted merge, for each distinct segmentby value we
10041080
* have to keep the corresponding latest batch open. Estimate the number of
@@ -1277,7 +1353,7 @@ ts_columnar_scan_generate_paths(PlannerInfo *root, RelOptInfo *chunk_rel, const
12771353
chunk_rel,
12781354
sort_info.needs_sequence_num);
12791355

1280-
if (sort_info.use_compressed_sort)
1356+
if (sort_info.use_compressed_sort || sort_info.use_batch_sorted_merge)
12811357
{
12821358
sort_info.required_compressed_pathkeys =
12831359
build_compressed_scan_pathkeys(&sort_info,
@@ -1492,6 +1568,8 @@ build_on_single_compressed_path(PlannerInfo *root, const Chunk *chunk, RelOptInf
14921568
* query here.
14931569
*/
14941570
path_copy->custom_path.path.pathkeys = sort_info->decompressed_sort_pathkeys;
1571+
path_copy->required_compressed_pathkeys = sort_info->required_compressed_pathkeys;
1572+
14951573
cost_batch_sorted_merge(root, compression_info, path_copy, compressed_path);
14961574

14971575
if (ts_guc_debug_require_batch_sorted_merge == DRO_Force)
@@ -2646,7 +2724,8 @@ create_compressed_scan_paths(PlannerInfo *root, RelOptInfo *compressed_rel,
26462724
}
26472725
}
26482726

2649-
if (sort_info->use_compressed_sort)
2727+
/* We can use sorted input before decompression in both cases */
2728+
if (sort_info->use_compressed_sort || sort_info->use_batch_sorted_merge)
26502729
{
26512730
/*
26522731
* If we can push down sort below decompression we temporarily switch

‎tsl/src/nodes/columnar_scan/exec.c‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ columnar_scan_state_create(CustomScan *cscan)
9898
Assert(list_length(chunk_state->decompression_map) ==
9999
list_length(chunk_state->is_segmentby_column));
100100

101+
chunk_state->done_fetching_batches = false;
101102
return (Node *) chunk_state;
102103
}
103104

@@ -201,6 +202,7 @@ columnar_scan_begin(CustomScanState *node, EState *estate, int eflags)
201202
Plan *compressed_scan = linitial(cscan->custom_plans);
202203
Assert(list_length(cscan->custom_plans) == 1);
203204

205+
chunk_state->done_fetching_batches = false;
204206
ts_stats_compression_acc_init(&dcontext->observ_acc);
205207

206208
PlanState *ps = &node->ss.ps;
@@ -451,12 +453,13 @@ columnar_scan_exec_impl(ColumnarScanState *chunk_state, const BatchQueueFunction
451453

452454
bqfuncs->pop(bq, dcontext);
453455

454-
while (bqfuncs->needs_next_batch(bq))
456+
while (!chunk_state->done_fetching_batches && bqfuncs->needs_next_batch(bq))
455457
{
456458
TupleTableSlot *subslot = ExecProcNode(linitial(chunk_state->csstate.custom_ps));
457459
if (TupIsNull(subslot))
458460
{
459461
/* Won't have more compressed tuples. */
462+
chunk_state->done_fetching_batches = true;
460463
break;
461464
}
462465

@@ -491,6 +494,7 @@ static void
491494
columnar_scan_rescan(CustomScanState *node)
492495
{
493496
ColumnarScanState *chunk_state = (ColumnarScanState *) node;
497+
chunk_state->done_fetching_batches = false;
494498
BatchQueue *bq = chunk_state->batch_queue;
495499

496500
bq->funcs->reset(bq);

0 commit comments

Comments
 (0)