Skip to content

Do not sort batches for Batch sorted merge over unordered chunks if query sort matches compressed sort #10303

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
natalya-aksman merged 1 commit into from
Jul 29, 2026
Merged

Do not sort batches for Batch sorted merge over unordered chunks if query sort matches compressed sort #10303

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
293 changes: 186 additions & 107 deletions tsl/src/nodes/columnar_scan/columnar_scan.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ build_compressed_scan_pathkeys(const SortInfo *sort_info, PlannerInfo *root, Lis
* If pathkeys contains non-segmentby columns the rest of the ordering
* requirements will be satisfied by ordering by sequence_num.
*/
if (sort_info->needs_sequence_num)
if (sort_info->needs_sequence_num || sort_info->use_batch_sorted_merge)
{
/* TODO: split up legacy sequence number path and non-sequence number path into dedicated
* functions. */
Expand Down Expand Up @@ -317,60 +317,10 @@ build_compressed_scan_pathkeys(const SortInfo *sort_info, PlannerInfo *root, Lis
column_name = get_attname(info->chunk_rte->relid, var->varattno, false);
int16 orderby_index = ts_array_position(info->settings->fd.orderby, column_name);
Assert(orderby_index != 0);
AttrNumber leading_attno;
AttrNumber trailing_attno;
orderby_sparse_metadata_attnos(info->settings,
info->compressed_rte->relid,
orderby_index,
&leading_attno,
&trailing_attno);

bool orderby_desc =
ts_array_get_element_bool(info->settings->fd.orderby_desc, orderby_index);

/*
* Compressed chunk indexes based on firstlast sparse indexes can have two
* orderings. New chunks index them as (first, last); chunks compressed before that
* change index a DESC column as (last, first). Only DESC columns can differ, so for
* those we check the chunk's index and follow whichever order it has.
*/
if (orderby_desc &&
orderby_sparse_kind(info->settings, orderby_index) == ORDERBY_SPARSE_FIRSTLAST)
{
orderby_firstlast_metadata_attnos(info->settings,
info->compressed_rte->relid,
orderby_index,
&leading_attno,
&trailing_attno);

ListCell *index_lc;
foreach (index_lc, info->compressed_rel->indexlist)
{
IndexOptInfo *index = lfirst(index_lc);
int leading_pos = -1;
int trailing_pos = -1;
for (int k = 0; k < index->nkeycolumns; k++)
{
if (index->indexkeys[k] == leading_attno)
{
leading_pos = k;
}
else if (index->indexkeys[k] == trailing_attno)
{
trailing_pos = k;
}
}
if (leading_pos >= 0 && trailing_pos >= 0)
{
if (trailing_pos < leading_pos)
{
AttrNumber tmp = leading_attno;
leading_attno = trailing_attno;
trailing_attno = tmp;
}
break;
}
}
}
bool orderby_nullsfirst =
ts_array_get_element_bool(info->settings->fd.orderby_nullsfirst, orderby_index);

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

Var *metadata_var = makeVar(info->compressed_rel->relid,
leading_attno,
var->vartype,
var->vartypmod,
var->varcollid,
var->varlevelsup);
Expr *leading_expr =
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
EquivalenceClass *leading_ec =
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
PathKey *leading_pk = make_canonical_pathkey(root,
leading_ec,
pk->pk_opfamily,
strategy,
nulls_first);
required_compressed_pathkeys = lappend(required_compressed_pathkeys, leading_pk);

metadata_var = makeVar(info->compressed_rel->relid,
trailing_attno,
var->vartype,
var->vartypmod,
var->varcollid,
var->varlevelsup);
Expr *trailing_expr =
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
EquivalenceClass *trailing_ec =
append_ec_for_metadata_col(root, info, trailing_expr, pk, opcintype);
PathKey *trailing_pk = make_canonical_pathkey(root,
trailing_ec,
pk->pk_opfamily,
strategy,
nulls_first);

required_compressed_pathkeys = lappend(required_compressed_pathkeys, trailing_pk);
/* For Batch sorted merge we need to sort on specially chosen leading attribute for
* each pathkey */
if (sort_info->use_batch_sorted_merge)
{
Oid sortop =
get_opfamily_member(pk->pk_opfamily, opcintype, opcintype, pk->pk_cmptype);
Oid opfamily, optype;
CompareType bsm_strategy;
if (!get_ordering_op_properties(sortop, &opfamily, &optype, &bsm_strategy))
{
elog(ERROR, "operator %u is not a valid ordering operator", sortop);
}
Assert(bsm_strategy == BTLessStrategyNumber ||
bsm_strategy == BTGreaterStrategyNumber);
char *leading_name;
char *trailing_name;
orderby_sparse_metadata_names(info->settings,
orderby_index,
&leading_name,
&trailing_name);
char *meta_col_name =
strategy == BTLessStrategyNumber ? leading_name : trailing_name;

AttrNumber attr_position =
get_attnum(info->compressed_rte->relid, meta_col_name);

if (attr_position == InvalidAttrNumber)
{
elog(ERROR, "couldn't find metadata column \"%s\"", meta_col_name);
}
Var *metadata_var = makeVar(info->compressed_rel->relid,
attr_position,
var->vartype,
var->vartypmod,
var->varcollid,
var->varlevelsup);
Expr *leading_expr =
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
EquivalenceClass *leading_ec =
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
PathKey *leading_pk = make_canonical_pathkey(root,
leading_ec,
pk->pk_opfamily,
strategy,
nulls_first);
required_compressed_pathkeys =
lappend(required_compressed_pathkeys, leading_pk);
}
/* Need to sort on compressed pathkeys matching compressed indexscan order */
else
{
AttrNumber leading_attno;
AttrNumber trailing_attno;
orderby_sparse_metadata_attnos(info->settings,
info->compressed_rte->relid,
orderby_index,
&leading_attno,
&trailing_attno);
/*
* Compressed chunk indexes based on firstlast sparse indexes can have two
* orderings. New chunks index them as (first, last); chunks compressed before
* that change index a DESC column as (last, first). Only DESC columns can
* differ, so for those we check the chunk's index and follow whichever order it
* has.
*/
if (orderby_desc && orderby_sparse_kind(info->settings, orderby_index) ==
ORDERBY_SPARSE_FIRSTLAST)
{
orderby_firstlast_metadata_attnos(info->settings,
info->compressed_rte->relid,
orderby_index,
&leading_attno,
&trailing_attno);

ListCell *index_lc;
foreach (index_lc, info->compressed_rel->indexlist)
{
IndexOptInfo *index = lfirst(index_lc);
int leading_pos = -1;
int trailing_pos = -1;
for (int k = 0; k < index->nkeycolumns; k++)
{
if (index->indexkeys[k] == leading_attno)
{
leading_pos = k;
}
else if (index->indexkeys[k] == trailing_attno)
{
trailing_pos = k;
}
}
if (leading_pos >= 0 && trailing_pos >= 0)
{
if (trailing_pos < leading_pos)
{
AttrNumber tmp = leading_attno;
leading_attno = trailing_attno;
trailing_attno = tmp;
}
break;
}
}
}
Var *metadata_var;
metadata_var = makeVar(info->compressed_rel->relid,
leading_attno,
var->vartype,
var->vartypmod,
var->varcollid,
var->varlevelsup);
Expr *leading_expr =
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
EquivalenceClass *leading_ec =
append_ec_for_metadata_col(root, info, leading_expr, pk, opcintype);
PathKey *leading_pk = make_canonical_pathkey(root,
leading_ec,
pk->pk_opfamily,
strategy,
nulls_first);
required_compressed_pathkeys =
lappend(required_compressed_pathkeys, leading_pk);

metadata_var = makeVar(info->compressed_rel->relid,
trailing_attno,
var->vartype,
var->vartypmod,
var->varcollid,
var->varlevelsup);
Expr *trailing_expr =
canonicalize_ec_expression((Expr *) metadata_var, opcintype, collation);
EquivalenceClass *trailing_ec =
append_ec_for_metadata_col(root, info, trailing_expr, pk, opcintype);
PathKey *trailing_pk = make_canonical_pathkey(root,
trailing_ec,
pk->pk_opfamily,
strategy,
nulls_first);

required_compressed_pathkeys =
lappend(required_compressed_pathkeys, trailing_pk);
}
}
}
}

return required_compressed_pathkeys;
}

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

/*
* Don't disable the compressed batch sorted merge plan with the enable_sort
* GUC. We have a separate GUC for it, and this way you can try to force the
* batch sorted merge plan by disabling sort.
*/
const bool old_enable_sort = enable_sort;
enable_sort = true;
cost_sort(&sort_path,
root,
dcpath->required_compressed_pathkeys,
/* We are utilizing compressed sort for batch sorted merge: do not need extra sort */
if (dcpath->required_compressed_pathkeys &&
pathkeys_contained_in(dcpath->required_compressed_pathkeys, compressed_path->pathkeys))
{
sort_path.rows = compressed_path->rows;
sort_path.startup_cost = compressed_path->startup_cost;
sort_path.total_cost = compressed_path->total_cost;
#if PG18_GE
compressed_path->disabled_nodes,
/* PG18 changes the way we handle disabled nodes so we
* need to take those into account as well.
*
* https://github.com/postgres/postgres/commit/e2225346
*/
sort_path.disabled_nodes = compressed_path->disabled_nodes;
#endif
compressed_path->total_cost,
compressed_path->rows,
compressed_path->pathtarget->width,
0.0,
work_mem,
-1);
enable_sort = old_enable_sort;

}
else
{
/*
* Don't disable the compressed batch sorted merge plan with the enable_sort
* GUC. We have a separate GUC for it, and this way you can try to force the
* batch sorted merge plan by disabling sort.
*/
const bool old_enable_sort = enable_sort;
enable_sort = true;
cost_sort(&sort_path,
root,
dcpath->required_compressed_pathkeys,
#if PG18_GE
compressed_path->disabled_nodes,
#endif
compressed_path->total_cost,
compressed_path->rows,
compressed_path->pathtarget->width,
0.0,
work_mem,
-1);
enable_sort = old_enable_sort;
}
/*
* In compressed batch sorted merge, for each distinct segmentby value we
* have to keep the corresponding latest batch open. Estimate the number of
Expand Down Expand Up @@ -1277,7 +1353,7 @@ ts_columnar_scan_generate_paths(PlannerInfo *root, RelOptInfo *chunk_rel, const
chunk_rel,
sort_info.needs_sequence_num);

if (sort_info.use_compressed_sort)
if (sort_info.use_compressed_sort || sort_info.use_batch_sorted_merge)
{
sort_info.required_compressed_pathkeys =
build_compressed_scan_pathkeys(&sort_info,
Expand Down Expand Up @@ -1492,6 +1568,8 @@ build_on_single_compressed_path(PlannerInfo *root, const Chunk *chunk, RelOptInf
* query here.
*/
path_copy->custom_path.path.pathkeys = sort_info->decompressed_sort_pathkeys;
path_copy->required_compressed_pathkeys = sort_info->required_compressed_pathkeys;

cost_batch_sorted_merge(root, compression_info, path_copy, compressed_path);

if (ts_guc_debug_require_batch_sorted_merge == DRO_Force)
Expand Down Expand Up @@ -2646,7 +2724,8 @@ create_compressed_scan_paths(PlannerInfo *root, RelOptInfo *compressed_rel,
}
}

if (sort_info->use_compressed_sort)
/* We can use sorted input before decompression in both cases */
if (sort_info->use_compressed_sort || sort_info->use_batch_sorted_merge)
{
/*
* If we can push down sort below decompression we temporarily switch
Expand Down
6 changes: 5 additions & 1 deletion tsl/src/nodes/columnar_scan/exec.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ columnar_scan_state_create(CustomScan *cscan)
Assert(list_length(chunk_state->decompression_map) ==
list_length(chunk_state->is_segmentby_column));

chunk_state->done_fetching_batches = false;
return (Node *) chunk_state;
}

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

chunk_state->done_fetching_batches = false;
ts_stats_compression_acc_init(&dcontext->observ_acc);

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

bqfuncs->pop(bq, dcontext);

while (bqfuncs->needs_next_batch(bq))
while (!chunk_state->done_fetching_batches && bqfuncs->needs_next_batch(bq))
{
TupleTableSlot *subslot = ExecProcNode(linitial(chunk_state->csstate.custom_ps));
if (TupIsNull(subslot))
{
/* Won't have more compressed tuples. */
chunk_state->done_fetching_batches = true;
break;
}

Expand Down Expand Up @@ -491,6 +494,7 @@ static void
columnar_scan_rescan(CustomScanState *node)
{
ColumnarScanState *chunk_state = (ColumnarScanState *) node;
chunk_state->done_fetching_batches = false;
BatchQueue *bq = chunk_state->batch_queue;

bq->funcs->reset(bq);
Expand Down
Loading
Loading