public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 4/4] Consider incremental sort paths in additional places
18+ messages / 6 participants
[nested] [flat]
* [PATCH 4/4] Consider incremental sort paths in additional places
@ 2019-07-28 13:59 Tomas Vondra <[email protected]>
0 siblings, 0 replies; 18+ messages in thread
From: Tomas Vondra @ 2019-07-28 13:59 UTC (permalink / raw)
---
contrib/postgres_fdw/postgres_fdw.c | 29 --
src/backend/optimizer/geqo/geqo_eval.c | 2 +-
src/backend/optimizer/path/allpaths.c | 208 +++++++++++++-
src/backend/optimizer/path/equivclass.c | 28 ++
src/backend/optimizer/plan/planner.c | 346 +++++++++++++++++++++++-
src/include/optimizer/paths.h | 3 +
6 files changed, 580 insertions(+), 36 deletions(-)
diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c
index 2175dff824..9fc53cad68 100644
--- a/contrib/postgres_fdw/postgres_fdw.c
+++ b/contrib/postgres_fdw/postgres_fdw.c
@@ -6523,35 +6523,6 @@ conversion_error_callback(void *arg)
}
}
-/*
- * Find an equivalence class member expression, all of whose Vars, come from
- * the indicated relation.
- */
-Expr *
-find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
-{
- ListCell *lc_em;
-
- foreach(lc_em, ec->ec_members)
- {
- EquivalenceMember *em = lfirst(lc_em);
-
- if (bms_is_subset(em->em_relids, rel->relids) &&
- !bms_is_empty(em->em_relids))
- {
- /*
- * If there is more than one equivalence member whose Vars are
- * taken entirely from this relation, we'll be content to choose
- * any one of those.
- */
- return em->em_expr;
- }
- }
-
- /* We didn't find any suitable equivalence class expression */
- return NULL;
-}
-
/*
* Find an equivalence class member expression to be computed as a sort column
* in the given target.
diff --git a/src/backend/optimizer/geqo/geqo_eval.c b/src/backend/optimizer/geqo/geqo_eval.c
index 6d897936d7..ff33acc7b6 100644
--- a/src/backend/optimizer/geqo/geqo_eval.c
+++ b/src/backend/optimizer/geqo/geqo_eval.c
@@ -274,7 +274,7 @@ merge_clump(PlannerInfo *root, List *clumps, Clump *new_clump, int num_gene,
* grouping_planner).
*/
if (old_clump->size + new_clump->size < num_gene)
- generate_gather_paths(root, joinrel, false);
+ generate_useful_gather_paths(root, joinrel, false);
/* Find and save the cheapest paths for this joinrel */
set_cheapest(joinrel);
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index ccf46dd0aa..93d967e812 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -556,7 +556,7 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
*/
if (rel->reloptkind == RELOPT_BASEREL &&
bms_membership(root->all_baserels) != BMS_SINGLETON)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Now find the cheapest of the paths for this rel */
set_cheapest(rel);
@@ -2727,6 +2727,210 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
}
}
+/*
+ * get_useful_pathkeys_for_relation
+ * Determine which orderings of a relation might be useful.
+ *
+ * Getting data in sorted order can be useful either because the requested
+ * order matches the final output ordering for the overall query we're
+ * planning, or because it enables an efficient merge join. Here, we try
+ * to figure out which pathkeys to consider.
+ *
+ * This allows us to do incremental sort on top of an index scan under a gather
+ * merge node, i.e. parallelized.
+ *
+ * XXX At the moment this can only ever return a list with a single element,
+ * because it looks at query_pathkeys only. So we might return the pathkeys
+ * directly, but it seems plausible we'll want to consider other orderings
+ * in the future. For example, we might want to consider pathkeys useful for
+ * merge joins.
+ */
+static List *
+get_useful_pathkeys_for_relation(PlannerInfo *root, RelOptInfo *rel)
+{
+ List *useful_pathkeys_list = NIL;
+
+ /*
+ * Considering query_pathkeys is always worth it, because it might allow us
+ * to avoid a total sort when we have a partially presorted path available.
+ */
+ if (root->query_pathkeys)
+ {
+ ListCell *lc;
+ List *pathkeys = NIL;
+
+ foreach(lc, root->query_pathkeys)
+ {
+ PathKey *pathkey = (PathKey *) lfirst(lc);
+ EquivalenceClass *pathkey_ec = pathkey->pk_eclass;
+
+ /*
+ * We can only build an Incremental Sort for pathkeys which contain
+ * an EC member in the current relation, so ignore any suffix of the
+ * list as soon as we find a pathkey without an EC member the
+ * relation.
+ *
+ * By still returning the prefix of the pathkeys list that does meet
+ * criteria of EC membership in the current relation, we enable not
+ * just an incremental sort on the entirety of query_pathkeys but
+ * also incremental sort below a JOIN.
+ */
+ if (!find_em_expr_for_rel(pathkey_ec, rel))
+ break;
+
+ pathkeys = lappend(pathkeys, pathkey);
+ }
+
+ if (pathkeys)
+ useful_pathkeys_list = lappend(useful_pathkeys_list, pathkeys);
+ }
+
+ return useful_pathkeys_list;
+}
+
+/*
+ * generate_useful_gather_paths
+ * Generate parallel access paths for a relation by pushing a Gather or
+ * Gather Merge on top of a partial path.
+ *
+ * Unlike plain generate_gather_paths, this looks both at pathkeys of input
+ * paths (aiming to preserve the ordering), but also considers ordering that
+ * might be useful for nodes above the gather merge node, and tries to add
+ * a sort (regular or incremental) to provide that.
+ */
+void
+generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
+{
+ ListCell *lc;
+ double rows;
+ double *rowsp = NULL;
+ List *useful_pathkeys_list = NIL;
+ Path *cheapest_partial_path = NULL;
+
+ /* If there are no partial paths, there's nothing to do here. */
+ if (rel->partial_pathlist == NIL)
+ return;
+
+ /* Should we override the rel's rowcount estimate? */
+ if (override_rows)
+ rowsp = &rows;
+
+ /* generate the regular gather (merge) paths */
+ generate_gather_paths(root, rel, override_rows);
+
+ /* consider incremental sort for interesting orderings */
+ useful_pathkeys_list = get_useful_pathkeys_for_relation(root, rel);
+
+ /* used for explicit (full) sort paths */
+ cheapest_partial_path = linitial(rel->partial_pathlist);
+
+ /*
+ * Consider incremental sort paths for each interesting ordering.
+ *
+ * XXX I wonder if we need to consider adding a projection here, as
+ * create_ordered_paths does.
+ */
+ foreach(lc, useful_pathkeys_list)
+ {
+ List *useful_pathkeys = lfirst(lc);
+ ListCell *lc2;
+ bool is_sorted;
+ int presorted_keys;
+
+ foreach(lc2, rel->partial_pathlist)
+ {
+ Path *subpath = (Path *) lfirst(lc2);
+ GatherMergePath *path;
+
+ /* path has no ordering at all, can't use incremental sort */
+ if (subpath->pathkeys == NIL)
+ continue;
+
+ is_sorted = pathkeys_common_contained_in(useful_pathkeys,
+ subpath->pathkeys,
+ &presorted_keys);
+
+ /*
+ * When the partial path is already sorted, we can just add a gather
+ * merge on top, and we're done - no point in adding explicit sort.
+ *
+ * XXX Can't we skip this (maybe only for the cheapest partial path)
+ * when the path is already sorted? Then it's likely duplicate with
+ * the path created by generate_gather_paths.
+ */
+ if (is_sorted)
+ {
+ path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
+ subpath->pathkeys, NULL, rowsp);
+
+ add_path(rel, &path->path);
+ continue;
+ }
+
+ Assert(!is_sorted);
+
+ /*
+ * Consider regular sort for the cheapest partial path (for each
+ * useful pathkeys). We know the path is not sorted, because we'd
+ * not get here otherwise.
+ *
+ * XXX This is not redundant with the gather merge path created in
+ * generate_gather_paths, because that merely preserves ordering of
+ * the cheapest partial path, while here we add an explicit sort to
+ * get match the useful ordering.
+ */
+ if (cheapest_partial_path == subpath)
+ {
+ Path *tmp;
+
+ tmp = (Path *) create_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ -1.0);
+
+ rows = tmp->rows * tmp->parallel_workers;
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+
+ /* Fall through */
+ }
+
+ /*
+ * Consider incremental sort, but only when the subpath is already
+ * partially sorted on a pathkey prefix.
+ */
+ if (enable_incrementalsort && presorted_keys > 0)
+ {
+ Path *tmp;
+
+ tmp = (Path *) create_incremental_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ presorted_keys,
+ -1);
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+ }
+ }
+ }
+}
+
/*
* make_rel_from_joinlist
* Build access paths using a "joinlist" to guide the join path search.
@@ -2899,7 +3103,7 @@ standard_join_search(PlannerInfo *root, int levels_needed, List *initial_rels)
* once we know the final targetlist (see grouping_planner).
*/
if (lev < levels_needed)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Find and save the cheapest paths for this rel */
set_cheapest(rel);
diff --git a/src/backend/optimizer/path/equivclass.c b/src/backend/optimizer/path/equivclass.c
index 4ef12547ee..b99cec00cb 100644
--- a/src/backend/optimizer/path/equivclass.c
+++ b/src/backend/optimizer/path/equivclass.c
@@ -774,6 +774,34 @@ get_eclass_for_sort_expr(PlannerInfo *root,
return newec;
}
+/*
+ * Find an equivalence class member expression, all of whose Vars, come from
+ * the indicated relation.
+ */
+Expr *
+find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
+{
+ ListCell *lc_em;
+
+ foreach(lc_em, ec->ec_members)
+ {
+ EquivalenceMember *em = lfirst(lc_em);
+
+ if (bms_is_subset(em->em_relids, rel->relids) &&
+ !bms_is_empty(em->em_relids))
+ {
+ /*
+ * If there is more than one equivalence member whose Vars are
+ * taken entirely from this relation, we'll be content to choose
+ * any one of those.
+ */
+ return em->em_expr;
+ }
+ }
+
+ /* We didn't find any suitable equivalence class expression */
+ return NULL;
+}
/*
* generate_base_implied_equalities
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 753e23676b..fb094e3be0 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -5077,6 +5077,67 @@ create_ordered_paths(PlannerInfo *root,
add_path(ordered_rel, path);
}
+
+ /*
+ * Consider incremental sort with a gather merge on partial paths.
+ *
+ * XXX This is probably duplicate with the paths we already generate
+ * in generate_useful_gather_paths in apply_scanjoin_target_to_paths.
+ */
+ if (enable_incrementalsort)
+ {
+ ListCell *lc;
+
+ foreach(lc, input_rel->partial_pathlist)
+ {
+ Path *input_path = (Path *) lfirst(lc);
+ Path *sorted_path = input_path;
+ bool is_sorted;
+ int presorted_keys;
+ double total_groups;
+
+ /*
+ * We don't care if this is the cheapest partial path - we can't
+ * simply skip it, because it may be partially sorted in which
+ * case we want to consider adding incremental sort (instead of
+ * full sort, which is what happens above).
+ */
+
+ is_sorted = pathkeys_common_contained_in(root->sort_pathkeys,
+ input_path->pathkeys,
+ &presorted_keys);
+
+ /* No point in adding incremental sort on fully sorted paths. */
+ if (is_sorted)
+ continue;
+
+ if (presorted_keys == 0)
+ continue;
+
+ /* Since we have presorted keys, consider incremental sort. */
+ sorted_path = (Path *) create_incremental_sort_path(root,
+ ordered_rel,
+ input_path,
+ root->sort_pathkeys,
+ presorted_keys,
+ limit_tuples);
+ total_groups = input_path->rows *
+ input_path->parallel_workers;
+ sorted_path = (Path *)
+ create_gather_merge_path(root, ordered_rel,
+ sorted_path,
+ sorted_path->pathtarget,
+ root->sort_pathkeys, NULL,
+ &total_groups);
+
+ /* Add projection step if needed */
+ if (sorted_path->pathtarget != target)
+ sorted_path = apply_projection_to_path(root, ordered_rel,
+ sorted_path, target);
+
+ add_path(ordered_rel, sorted_path);
+ }
+ }
}
/*
@@ -6431,7 +6492,9 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
foreach(lc, input_rel->pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6490,6 +6553,80 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
Assert(false);
}
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have added Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, no point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ /* Now decide what to stick atop it */
+ if (parse->groupingSets)
+ {
+ consider_groupingsets_paths(root, grouped_rel,
+ path, true, can_hash,
+ gd, agg_costs, dNumGroups);
+ }
+ else if (parse->hasAggs)
+ {
+ /*
+ * We have aggregation, possibly with plain GROUP BY. Make
+ * an AggPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_agg_path(root,
+ grouped_rel,
+ path,
+ grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_SIMPLE,
+ parse->groupClause,
+ havingQual,
+ agg_costs,
+ dNumGroups));
+ }
+ else if (parse->groupClause)
+ {
+ /*
+ * We have GROUP BY without aggregation or grouping sets.
+ * Make a GroupPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_group_path(root,
+ grouped_rel,
+ path,
+ parse->groupClause,
+ havingQual,
+ dNumGroups));
+ }
+ else
+ {
+ /* Other cases should have been handled above */
+ Assert(false);
+ }
}
/*
@@ -6501,12 +6638,18 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
foreach(lc, partially_grouped_rel->pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
+ bool is_sorted;
+ int presorted_keys;
+
+ is_sorted = pathkeys_contained_in(root->group_pathkeys,
+ path->pathkeys);
/*
* Insert a Sort node, if required. But there's no point in
* sorting anything but the cheapest path.
*/
- if (!pathkeys_contained_in(root->group_pathkeys, path->pathkeys))
+ if (!is_sorted)
{
if (path != partially_grouped_rel->cheapest_total_path)
continue;
@@ -6537,6 +6680,56 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
parse->groupClause,
havingQual,
dNumGroups));
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental
+ * sort is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have added Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ if (parse->hasAggs)
+ add_path(grouped_rel, (Path *)
+ create_agg_path(root,
+ grouped_rel,
+ path,
+ grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_FINAL_DESERIAL,
+ parse->groupClause,
+ havingQual,
+ agg_final_costs,
+ dNumGroups));
+ else
+ add_path(grouped_rel, (Path *)
+ create_group_path(root,
+ grouped_rel,
+ path,
+ parse->groupClause,
+ havingQual,
+ dNumGroups));
}
}
}
@@ -6808,6 +7001,58 @@ create_partial_grouping_paths(PlannerInfo *root,
dNumPartialGroups));
}
}
+
+ /* Consider incremental sort on all partial paths, if enabled. */
+ if (enable_incrementalsort)
+ {
+ foreach(lc, input_rel->pathlist)
+ {
+ Path *path = (Path *) lfirst(lc);
+ bool is_sorted;
+ int presorted_keys;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* Ignore already sorted paths */
+ if (is_sorted)
+ continue;
+
+ if (presorted_keys == 0)
+ continue;
+
+ /* Since we have presorted keys, consider incremental sort. */
+ path = (Path *) create_incremental_sort_path(root,
+ partially_grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ if (parse->hasAggs)
+ add_path(partially_grouped_rel, (Path *)
+ create_agg_path(root,
+ partially_grouped_rel,
+ path,
+ partially_grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_INITIAL_SERIAL,
+ parse->groupClause,
+ NIL,
+ agg_partial_costs,
+ dNumPartialGroups));
+ else
+ add_path(partially_grouped_rel, (Path *)
+ create_group_path(root,
+ partially_grouped_rel,
+ path,
+ parse->groupClause,
+ NIL,
+ dNumPartialGroups));
+ }
+ }
+
}
if (can_sort && cheapest_partial_path != NULL)
@@ -6816,7 +7061,9 @@ create_partial_grouping_paths(PlannerInfo *root,
foreach(lc, input_rel->partial_pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6851,6 +7098,56 @@ create_partial_grouping_paths(PlannerInfo *root,
NIL,
dNumPartialPartialGroups));
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have added Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ partially_grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ if (parse->hasAggs)
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_agg_path(root,
+ partially_grouped_rel,
+ path,
+ partially_grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_INITIAL_SERIAL,
+ parse->groupClause,
+ NIL,
+ agg_partial_costs,
+ dNumPartialPartialGroups));
+ else
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_group_path(root,
+ partially_grouped_rel,
+ path,
+ parse->groupClause,
+ NIL,
+ dNumPartialPartialGroups));
}
}
@@ -6948,10 +7245,11 @@ create_partial_grouping_paths(PlannerInfo *root,
static void
gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
{
+ ListCell *lc;
Path *cheapest_partial_path;
/* Try Gather for unordered paths and Gather Merge for ordered ones. */
- generate_gather_paths(root, rel, true);
+ generate_useful_gather_paths(root, rel, true);
/* Try cheapest partial path + explicit Sort + Gather Merge. */
cheapest_partial_path = linitial(rel->partial_pathlist);
@@ -6977,6 +7275,46 @@ gather_grouping_paths(PlannerInfo *root, RelOptInfo *rel)
add_path(rel, path);
}
+
+ if (!enable_incrementalsort)
+ return;
+
+ /* also consider incremental sort on partial paths, if enabled */
+ foreach(lc, rel->partial_pathlist)
+ {
+ Path *path = (Path *) lfirst(lc);
+ bool is_sorted;
+ int presorted_keys;
+ double total_groups;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ if (is_sorted)
+ continue;
+
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ path = (Path *)
+ create_gather_merge_path(root,
+ rel,
+ path,
+ rel->reltarget,
+ root->group_pathkeys,
+ NULL,
+ &total_groups);
+
+ add_path(rel, path);
+ }
}
/*
@@ -7078,7 +7416,7 @@ apply_scanjoin_target_to_paths(PlannerInfo *root,
* paths by doing it after the final scan/join target has been
* applied.
*/
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Can't use parallel query above this level. */
rel->partial_pathlist = NIL;
@@ -7232,7 +7570,7 @@ apply_scanjoin_target_to_paths(PlannerInfo *root,
* one of the generated paths may turn out to be the cheapest one.
*/
if (rel->consider_parallel && !IS_OTHER_REL(rel))
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/*
* Reassess which paths are the cheapest, now that we've potentially added
diff --git a/src/include/optimizer/paths.h b/src/include/optimizer/paths.h
index 85f5fe37ea..665f4065a4 100644
--- a/src/include/optimizer/paths.h
+++ b/src/include/optimizer/paths.h
@@ -54,6 +54,8 @@ extern RelOptInfo *standard_join_search(PlannerInfo *root, int levels_needed,
extern void generate_gather_paths(PlannerInfo *root, RelOptInfo *rel,
bool override_rows);
+extern void generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel,
+ bool override_rows);
extern int compute_parallel_worker(RelOptInfo *rel, double heap_pages,
double index_pages, int max_workers);
extern void create_partial_bitmap_paths(PlannerInfo *root, RelOptInfo *rel,
@@ -135,6 +137,7 @@ extern EquivalenceClass *get_eclass_for_sort_expr(PlannerInfo *root,
Index sortref,
Relids rel,
bool create_it);
+extern Expr *find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel);
extern void generate_base_implied_equalities(PlannerInfo *root);
extern List *generate_join_implied_equalities(PlannerInfo *root,
Relids join_relids,
--
2.21.1
--lxi422ive4a7neeu--
^ permalink raw reply [nested|flat] 18+ messages in thread
* [PATCH 5/8] Consider incremental sort paths in additional places
@ 2019-07-28 13:59 Tomas Vondra <[email protected]>
0 siblings, 0 replies; 18+ messages in thread
From: Tomas Vondra @ 2019-07-28 13:59 UTC (permalink / raw)
---
src/backend/optimizer/path/allpaths.c | 221 +++++++++++++++++++++++++-
src/backend/optimizer/plan/planner.c | 130 ++++++++++++++-
src/include/optimizer/paths.h | 2 +
3 files changed, 350 insertions(+), 3 deletions(-)
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index ccf46dd0aa..9a92948fe3 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -556,7 +556,7 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
*/
if (rel->reloptkind == RELOPT_BASEREL &&
bms_membership(root->all_baserels) != BMS_SINGLETON)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Now find the cheapest of the paths for this rel */
set_cheapest(rel);
@@ -2727,6 +2727,223 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
}
}
+/*
+ * Find an equivalence class member expression, all of whose Vars, come from
+ * the indicated relation.
+ */
+static Expr *
+find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
+{
+ ListCell *lc_em;
+
+ foreach(lc_em, ec->ec_members)
+ {
+ EquivalenceMember *em = lfirst(lc_em);
+
+ if (bms_is_subset(em->em_relids, rel->relids) &&
+ !bms_is_empty(em->em_relids))
+ {
+ /*
+ * If there is more than one equivalence member whose Vars are
+ * taken entirely from this relation, we'll be content to choose
+ * any one of those.
+ */
+ return em->em_expr;
+ }
+ }
+
+ /* We didn't find any suitable equivalence class expression */
+ return NULL;
+}
+
+/*
+ * get_useful_pathkeys_for_relation
+ * Determine which orderings of a relation might be useful.
+ *
+ * Getting data in sorted order can be useful either because the requested
+ * order matches the final output ordering for the overall query we're
+ * planning, or because it enables an efficient merge join. Here, we try
+ * to figure out which pathkeys to consider.
+ */
+static List *
+get_useful_pathkeys_for_relation(PlannerInfo *root, RelOptInfo *rel)
+{
+ List *useful_pathkeys_list = NIL;
+ ListCell *lc;
+
+ /*
+ * Pushing the query_pathkeys to the remote server is always worth
+ * considering, because it might let us avoid a local sort.
+ */
+ if (root->query_pathkeys)
+ {
+ bool query_pathkeys_ok = true;
+
+ foreach(lc, root->query_pathkeys)
+ {
+ PathKey *pathkey = (PathKey *) lfirst(lc);
+ EquivalenceClass *pathkey_ec = pathkey->pk_eclass;
+ Expr *em_expr;
+
+ /*
+ * The planner and executor don't have any clever strategy for
+ * taking data sorted by a prefix of the query's pathkeys and
+ * getting it to be sorted by all of those pathkeys. We'll just
+ * end up re-sorting the entire data set. So, unless we can push
+ * down all of the query pathkeys, forget it.
+ *
+ * is_foreign_expr would detect volatile expressions as well, but
+ * checking ec_has_volatile here saves some cycles.
+ */
+ if (pathkey_ec->ec_has_volatile ||
+ !(em_expr = find_em_expr_for_rel(pathkey_ec, rel)))
+ {
+ query_pathkeys_ok = false;
+ break;
+ }
+ }
+
+ /*
+ * This ends up allowing us to do incremental sort on top of an index
+ * scan all parallelized under a gather merge node.
+ */
+ if (query_pathkeys_ok)
+ useful_pathkeys_list = list_make1(list_copy(root->query_pathkeys));
+ }
+
+ return useful_pathkeys_list;
+}
+
+/*
+ * generate_useful_gather_paths
+ * Generate parallel access paths for a relation by pushing a Gather or
+ * Gather Merge on top of a partial path.
+ *
+ * Unlike generate_gather_paths, this does not look only at pathkeys of the
+ * input paths (aiming to preserve the ordering). It also considers ordering
+ * that might be useful by nodes above the gather merge node, and tries to
+ * add a sort (regular or incremental) to provide that.
+ */
+void
+generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
+{
+ ListCell *lc;
+ double rows;
+ double *rowsp = NULL;
+ List *useful_pathkeys_list = NIL;
+ Path *cheapest_partial_path = NULL;
+
+ /* If there are no partial paths, there's nothing to do here. */
+ if (rel->partial_pathlist == NIL)
+ return;
+
+ /* Should we override the rel's rowcount estimate? */
+ if (override_rows)
+ rowsp = &rows;
+
+ /* generate the regular gather merge paths */
+ generate_gather_paths(root, rel, override_rows);
+
+ /* when incremental sort is disabled, we're done */
+ if (!enable_incrementalsort)
+ return;
+
+ /* consider incremental sort for interesting orderings */
+ useful_pathkeys_list = get_useful_pathkeys_for_relation(root, rel);
+
+ /* used for explicit sort paths */
+ cheapest_partial_path = linitial(rel->partial_pathlist);
+
+ /*
+ * Consider incremental sort paths for each interesting ordering.
+ *
+ * XXX I wonder if we need to consider adding a projection here, as
+ * create_ordered_paths does.
+ */
+ foreach(lc, useful_pathkeys_list)
+ {
+ List *useful_pathkeys = lfirst(lc);
+ ListCell *lc2;
+ bool is_sorted;
+ int presorted_keys;
+
+ foreach(lc2, rel->partial_pathlist)
+ {
+ Path *subpath = (Path *) lfirst(lc2);
+ GatherMergePath *path;
+
+ /* path has no ordering at all, can't use incremental sort */
+ if (subpath->pathkeys == NIL)
+ continue;
+
+ is_sorted = pathkeys_common_contained_in(useful_pathkeys,
+ subpath->pathkeys,
+ &presorted_keys);
+
+ if (is_sorted)
+ {
+ path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
+ subpath->pathkeys, NULL, rowsp);
+
+ add_path(rel, &path->path);
+ continue;
+ }
+
+ Assert(!is_sorted);
+
+ /*
+ * consider regular sort for cheapest partial path (for each
+ * useful pathkeys)
+ */
+ if (cheapest_partial_path == subpath)
+ {
+ Path *tmp;
+
+ tmp = (Path *) create_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ -1.0);
+
+ rows = tmp->rows * tmp->parallel_workers;
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+
+ /* Fall through */
+ }
+
+ /* Also consider incremental sort */
+ if (presorted_keys > 0)
+ {
+ Path *tmp;
+
+ tmp = (Path *) create_incremental_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ presorted_keys,
+ -1);
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+ }
+ }
+ }
+}
+
/*
* make_rel_from_joinlist
* Build access paths using a "joinlist" to guide the join path search.
@@ -2899,7 +3116,7 @@ standard_join_search(PlannerInfo *root, int levels_needed, List *initial_rels)
* once we know the final targetlist (see grouping_planner).
*/
if (lev < levels_needed)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Find and save the cheapest paths for this rel */
set_cheapest(rel);
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index 0e01cf8cb1..46dc355af3 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -6434,7 +6434,9 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
foreach(lc, input_rel->pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6493,6 +6495,80 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
Assert(false);
}
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have added Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ /* Now decide what to stick atop it */
+ if (parse->groupingSets)
+ {
+ consider_groupingsets_paths(root, grouped_rel,
+ path, true, can_hash,
+ gd, agg_costs, dNumGroups);
+ }
+ else if (parse->hasAggs)
+ {
+ /*
+ * We have aggregation, possibly with plain GROUP BY. Make
+ * an AggPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_agg_path(root,
+ grouped_rel,
+ path,
+ grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_SIMPLE,
+ parse->groupClause,
+ havingQual,
+ agg_costs,
+ dNumGroups));
+ }
+ else if (parse->groupClause)
+ {
+ /*
+ * We have GROUP BY without aggregation or grouping sets.
+ * Make a GroupPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_group_path(root,
+ grouped_rel,
+ path,
+ parse->groupClause,
+ havingQual,
+ dNumGroups));
+ }
+ else
+ {
+ /* Other cases should have been handled above */
+ Assert(false);
+ }
}
/*
@@ -6819,7 +6895,9 @@ create_partial_grouping_paths(PlannerInfo *root,
foreach(lc, input_rel->partial_pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6854,6 +6932,56 @@ create_partial_grouping_paths(PlannerInfo *root,
NIL,
dNumPartialPartialGroups));
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have added Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ partially_grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ if (parse->hasAggs)
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_agg_path(root,
+ partially_grouped_rel,
+ path,
+ partially_grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_INITIAL_SERIAL,
+ parse->groupClause,
+ NIL,
+ agg_partial_costs,
+ dNumPartialPartialGroups));
+ else
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_group_path(root,
+ partially_grouped_rel,
+ path,
+ parse->groupClause,
+ NIL,
+ dNumPartialPartialGroups));
}
}
@@ -7235,7 +7363,7 @@ apply_scanjoin_target_to_paths(PlannerInfo *root,
* one of the generated paths may turn out to be the cheapest one.
*/
if (rel->consider_parallel && !IS_OTHER_REL(rel))
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/*
* Reassess which paths are the cheapest, now that we've potentially added
diff --git a/src/include/optimizer/paths.h b/src/include/optimizer/paths.h
index 85f5fe37ea..f6994779de 100644
--- a/src/include/optimizer/paths.h
+++ b/src/include/optimizer/paths.h
@@ -54,6 +54,8 @@ extern RelOptInfo *standard_join_search(PlannerInfo *root, int levels_needed,
extern void generate_gather_paths(PlannerInfo *root, RelOptInfo *rel,
bool override_rows);
+extern void generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel,
+ bool override_rows);
extern int compute_parallel_worker(RelOptInfo *rel, double heap_pages,
double index_pages, int max_workers);
extern void create_partial_bitmap_paths(PlannerInfo *root, RelOptInfo *rel,
--
2.21.1
--dfcjsgdukgytabqd
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment; filename="v39-0006-fix.patch"
^ permalink raw reply [nested|flat] 18+ messages in thread
* [PATCH 4/5] Consider incremental sort paths in additional places
@ 2019-07-28 13:59 Tomas Vondra <[email protected]>
0 siblings, 0 replies; 18+ messages in thread
From: Tomas Vondra @ 2019-07-28 13:59 UTC (permalink / raw)
---
src/backend/optimizer/path/allpaths.c | 222 +++++++++++++++++++++++++-
src/backend/optimizer/plan/planner.c | 130 ++++++++++++++-
src/include/optimizer/paths.h | 2 +
3 files changed, 351 insertions(+), 3 deletions(-)
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index ccf46dd0aa..8d9c25e18f 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -556,7 +556,7 @@ set_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
*/
if (rel->reloptkind == RELOPT_BASEREL &&
bms_membership(root->all_baserels) != BMS_SINGLETON)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Now find the cheapest of the paths for this rel */
set_cheapest(rel);
@@ -2727,6 +2727,224 @@ generate_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
}
}
+/*
+ * Find an equivalence class member expression, all of whose Vars, come from
+ * the indicated relation.
+ */
+static Expr *
+find_em_expr_for_rel(EquivalenceClass *ec, RelOptInfo *rel)
+{
+ ListCell *lc_em;
+
+ foreach(lc_em, ec->ec_members)
+ {
+ EquivalenceMember *em = lfirst(lc_em);
+
+ if (bms_is_subset(em->em_relids, rel->relids) &&
+ !bms_is_empty(em->em_relids))
+ {
+ /*
+ * If there is more than one equivalence member whose Vars are
+ * taken entirely from this relation, we'll be content to choose
+ * any one of those.
+ */
+ return em->em_expr;
+ }
+ }
+
+ /* We didn't find any suitable equivalence class expression */
+ return NULL;
+}
+
+/*
+ * get_useful_pathkeys_for_relation
+ * Determine which orderings of a relation might be useful.
+ *
+ * Getting data in sorted order can be useful either because the requested
+ * order matches the final output ordering for the overall query we're
+ * planning, or because it enables an efficient merge join. Here, we try
+ * to figure out which pathkeys to consider.
+ */
+static List *
+get_useful_pathkeys_for_relation(PlannerInfo *root, RelOptInfo *rel)
+{
+ List *useful_pathkeys_list = NIL;
+ ListCell *lc;
+
+ /*
+ * Pushing the query_pathkeys to the remote server is always worth
+ * considering, because it might let us avoid a local sort.
+ */
+ if (root->query_pathkeys)
+ {
+ bool query_pathkeys_ok = true;
+
+ foreach(lc, root->query_pathkeys)
+ {
+ PathKey *pathkey = (PathKey *) lfirst(lc);
+ EquivalenceClass *pathkey_ec = pathkey->pk_eclass;
+ Expr *em_expr;
+
+ /*
+ * The planner and executor don't have any clever strategy for
+ * taking data sorted by a prefix of the query's pathkeys and
+ * getting it to be sorted by all of those pathkeys. We'll just
+ * end up resorting the entire data set. So, unless we can push
+ * down all of the query pathkeys, forget it.
+ *
+ * is_foreign_expr would detect volatile expressions as well, but
+ * checking ec_has_volatile here saves some cycles.
+ */
+ if (pathkey_ec->ec_has_volatile ||
+ !(em_expr = find_em_expr_for_rel(pathkey_ec, rel)))
+ {
+ query_pathkeys_ok = false;
+ break;
+ }
+ }
+
+ /*
+ * This ends up allowing us to do incremental sort on top of
+ * an index scan all parallelized under a gather merge node.
+ */
+ if (query_pathkeys_ok)
+ useful_pathkeys_list = list_make1(list_copy(root->query_pathkeys));
+ }
+
+ return useful_pathkeys_list;
+}
+
+/*
+ * generate_useful_gather_paths
+ * Generate parallel access paths for a relation by pushing a Gather or
+ * Gather Merge on top of a partial path.
+ *
+ * Unlike generate_gather_paths, this does not look just as pathkeys of the
+ * input paths (aiming to preserve the ordering). It also considers ordering
+ * that might be useful by nodes above the gather merge node, and tries to
+ * add a sort (regular or incremental) to provide that.
+ */
+void
+generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel, bool override_rows)
+{
+ ListCell *lc;
+ double rows;
+ double *rowsp = NULL;
+ List *useful_pathkeys_list = NIL;
+ Path *cheapest_partial_path = NULL;
+
+ /* If there are no partial paths, there's nothing to do here. */
+ if (rel->partial_pathlist == NIL)
+ return;
+
+ /* Should we override the rel's rowcount estimate? */
+ if (override_rows)
+ rowsp = &rows;
+
+ /* generate the regular gather merge paths */
+ generate_gather_paths(root, rel, override_rows);
+
+ /* when incremental sort is disabled, we're done */
+ if (!enable_incrementalsort)
+ return;
+
+ /* consider incremental sort for interesting orderings */
+ useful_pathkeys_list = get_useful_pathkeys_for_relation(root, rel);
+
+ /* used for explicit sort paths */
+ cheapest_partial_path = linitial(rel->partial_pathlist);
+
+ /*
+ * Consider incremental sort paths for each interesting ordering.
+ *
+ * XXX I wonder if we need to consider adding a projection here, as
+ * create_ordered_paths does.
+ */
+ foreach(lc, useful_pathkeys_list)
+ {
+ List *useful_pathkeys = lfirst(lc);
+ ListCell *lc2;
+ bool is_sorted;
+ int presorted_keys;
+
+ foreach(lc2, rel->partial_pathlist)
+ {
+ Path *subpath = (Path *) lfirst(lc2);
+ GatherMergePath *path;
+
+ /* path has no ordering at all, can't use incremental sort */
+ if (subpath->pathkeys == NIL)
+ continue;
+
+ is_sorted = pathkeys_common_contained_in(useful_pathkeys,
+ subpath->pathkeys,
+ &presorted_keys);
+
+ if (is_sorted)
+ {
+ path = create_gather_merge_path(root, rel, subpath, rel->reltarget,
+ subpath->pathkeys, NULL, rowsp);
+
+ add_path(rel, &path->path);
+ continue;
+ }
+
+ /* now we know is_sorted == false */
+
+ /*
+ * consider regular sort for cheapest partial path (for each
+ * useful pathkeys)
+ */
+ if (cheapest_partial_path == subpath)
+ {
+ Path *tmp;
+
+ tmp = (Path *) create_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ -1.0);
+
+ rows = tmp->rows * tmp->parallel_workers;
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+
+ /* continue */
+ }
+
+ /* finally, consider incremental sort */
+ if (presorted_keys > 0)
+ {
+ Path *tmp;
+
+ /* Also consider incremental sort. */
+ tmp = (Path *) create_incremental_sort_path(root,
+ rel,
+ subpath,
+ useful_pathkeys,
+ presorted_keys,
+ -1);
+
+ path = create_gather_merge_path(root, rel,
+ tmp,
+ rel->reltarget,
+ tmp->pathkeys,
+ NULL,
+ rowsp);
+
+ add_path(rel, &path->path);
+ }
+ }
+ }
+}
+
/*
* make_rel_from_joinlist
* Build access paths using a "joinlist" to guide the join path search.
@@ -2899,7 +3117,7 @@ standard_join_search(PlannerInfo *root, int levels_needed, List *initial_rels)
* once we know the final targetlist (see grouping_planner).
*/
if (lev < levels_needed)
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/* Find and save the cheapest paths for this rel */
set_cheapest(rel);
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index e9918ffcb4..84ed69ec5e 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -6424,7 +6424,9 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
foreach(lc, input_rel->pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6483,6 +6485,80 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel,
Assert(false);
}
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have addes Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ /* Now decide what to stick atop it */
+ if (parse->groupingSets)
+ {
+ consider_groupingsets_paths(root, grouped_rel,
+ path, true, can_hash,
+ gd, agg_costs, dNumGroups);
+ }
+ else if (parse->hasAggs)
+ {
+ /*
+ * We have aggregation, possibly with plain GROUP BY. Make
+ * an AggPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_agg_path(root,
+ grouped_rel,
+ path,
+ grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_SIMPLE,
+ parse->groupClause,
+ havingQual,
+ agg_costs,
+ dNumGroups));
+ }
+ else if (parse->groupClause)
+ {
+ /*
+ * We have GROUP BY without aggregation or grouping sets.
+ * Make a GroupPath.
+ */
+ add_path(grouped_rel, (Path *)
+ create_group_path(root,
+ grouped_rel,
+ path,
+ parse->groupClause,
+ havingQual,
+ dNumGroups));
+ }
+ else
+ {
+ /* Other cases should have been handled above */
+ Assert(false);
+ }
}
/*
@@ -6807,7 +6883,9 @@ create_partial_grouping_paths(PlannerInfo *root,
foreach(lc, input_rel->partial_pathlist)
{
Path *path = (Path *) lfirst(lc);
+ Path *path_original = path;
bool is_sorted;
+ int presorted_keys;
is_sorted = pathkeys_contained_in(root->group_pathkeys,
path->pathkeys);
@@ -6842,6 +6920,56 @@ create_partial_grouping_paths(PlannerInfo *root,
NIL,
dNumPartialPartialGroups));
}
+
+ /*
+ * Now we may consider incremental sort on this path, but only
+ * when the path is not already sorted and when incremental sort
+ * is enabled.
+ */
+ if (is_sorted || !enable_incrementalsort)
+ continue;
+
+ /* Restore the input path (we might have addes Sort on top). */
+ path = path_original;
+
+ is_sorted = pathkeys_common_contained_in(root->group_pathkeys,
+ path->pathkeys,
+ &presorted_keys);
+
+ /* We've already skipped fully sorted paths above. */
+ Assert(!is_sorted);
+
+ /* no shared prefix, not point in building incremental sort */
+ if (presorted_keys == 0)
+ continue;
+
+ path = (Path *) create_incremental_sort_path(root,
+ partially_grouped_rel,
+ path,
+ root->group_pathkeys,
+ presorted_keys,
+ -1.0);
+
+ if (parse->hasAggs)
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_agg_path(root,
+ partially_grouped_rel,
+ path,
+ partially_grouped_rel->reltarget,
+ parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_INITIAL_SERIAL,
+ parse->groupClause,
+ NIL,
+ agg_partial_costs,
+ dNumPartialPartialGroups));
+ else
+ add_partial_path(partially_grouped_rel, (Path *)
+ create_group_path(root,
+ partially_grouped_rel,
+ path,
+ parse->groupClause,
+ NIL,
+ dNumPartialPartialGroups));
}
}
@@ -7223,7 +7351,7 @@ apply_scanjoin_target_to_paths(PlannerInfo *root,
* one of the generated paths may turn out to be the cheapest one.
*/
if (rel->consider_parallel && !IS_OTHER_REL(rel))
- generate_gather_paths(root, rel, false);
+ generate_useful_gather_paths(root, rel, false);
/*
* Reassess which paths are the cheapest, now that we've potentially added
diff --git a/src/include/optimizer/paths.h b/src/include/optimizer/paths.h
index 77c03149cd..d778b884a9 100644
--- a/src/include/optimizer/paths.h
+++ b/src/include/optimizer/paths.h
@@ -54,6 +54,8 @@ extern RelOptInfo *standard_join_search(PlannerInfo *root, int levels_needed,
extern void generate_gather_paths(PlannerInfo *root, RelOptInfo *rel,
bool override_rows);
+extern void generate_useful_gather_paths(PlannerInfo *root, RelOptInfo *rel,
+ bool override_rows);
extern int compute_parallel_worker(RelOptInfo *rel, double heap_pages,
double index_pages, int max_workers);
extern void create_partial_bitmap_paths(PlannerInfo *root, RelOptInfo *rel,
--
2.21.1
--xvcuvgto6w2bcqjv
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment;
filename="0005-A-couple-more-places-for-incremental-sort-20200310.patch"
^ permalink raw reply [nested|flat] 18+ messages in thread
* Allow logical replication to copy tables in binary format
@ 2022-08-10 15:03 Melih Mutlu <[email protected]>
0 siblings, 2 replies; 18+ messages in thread
From: Melih Mutlu @ 2022-08-10 15:03 UTC (permalink / raw)
To: pgsql-hackers
Hey hackers,
I see that logical replication subscriptions have an option to enable
binary [1].
When it's enabled, subscription requests publisher to send data in binary
format.
But this is only the case for apply phase. In tablesync, tables are still
copied as text.
To copy tables, COPY command is used and that command supports copying in
binary. So it seemed to me possible to copy in binary for tablesync too.
I'm not sure if there is a reason to always copy tables in text format. But
I couldn't see why not to do it in binary if it's enabled.
You can find the small patch that only enables binary copy attached.
What do you think about this change? Does it make sense? Am I missing
something?
[1] https://www.postgresql.org/docs/15/sql-createsubscription.html
Best,
Melih
Attachments:
[application/octet-stream] 0001-Allow-logical-replication-to-copy-table-in-binary.patch (1.9K, ../../CAGPVpCQvAziCLknEnygY0v1-KBtg+Om-9JHJYZOnNPKFJPompw@mail.gmail.com/3-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From 3d4cf3aaeb6bf341634743da342347408bd3752e Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH] Allow logical replication to copy table in binary
If binary option is enabled in a subscription, copy tables in binary
format during table synchronization.
---
src/backend/replication/logical/tablesync.c | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6a01ffd273..a3b128f6fb 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -101,6 +101,7 @@
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
+#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
@@ -1032,6 +1033,7 @@ copy_table(Relation rel)
CopyFromState cstate;
List *attnamelist;
ParseState *pstate;
+ List *options = NIL;
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
@@ -1110,6 +1112,13 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ if (MySubscription->binary)
+ {
+ appendStringInfoString(&cmd, " WITH (FORMAT binary)");
+ options = lappend(options, makeDefElem("format", (Node *) makeString("binary"), -1));
+ }
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1126,7 +1135,7 @@ copy_table(Relation rel)
NULL, false, false);
attnamelist = make_copy_attnamelist(relmapentry);
- cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
+ cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
/* Do the copy */
(void) CopyFrom(cstate);
--
2.25.1
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-11 02:03 Euler Taveira <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 1 reply; 18+ messages in thread
From: Euler Taveira @ 2022-08-11 02:03 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; pgsql-hackers
On Wed, Aug 10, 2022, at 12:03 PM, Melih Mutlu wrote:
> I see that logical replication subscriptions have an option to enable binary [1].
> When it's enabled, subscription requests publisher to send data in binary format.
> But this is only the case for apply phase. In tablesync, tables are still copied as text.
This option could have been included in the commit 9de77b54531; it wasn't.
Maybe it wasn't considered because the initial table synchronization can be a
separate step in your logical replication setup idk. I agree that the binary
option should be available for the initial table synchronization.
> To copy tables, COPY command is used and that command supports copying in binary. So it seemed to me possible to copy in binary for tablesync too.
> I'm not sure if there is a reason to always copy tables in text format. But I couldn't see why not to do it in binary if it's enabled.
The reason to use text format is that it is error prone. There are restrictions
while using the binary format. For example, if your schema has different data
types for a certain column, the copy will fail. Even with such restrictions, I
think it is worth adding it.
> You can find the small patch that only enables binary copy attached.
I have a few points about your implementation.
* Are we considering to support prior Postgres versions too? These releases
support binary mode but it could be an unexpected behavior (initial sync in
binary mode) for a publisher using 14 or 15 and a subscriber using 16. IMO
you should only allow it for publisher on 16 or later.
* Docs should say that the binary option also applies to initial table
synchronization and possibly emphasize some of the restrictions.
* Tests. Are the current tests enough? 014_binary.pl.
--
Euler Taveira
EDB https://www.enterprisedb.com/
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-11 11:04 Amit Kapila <[email protected]>
parent: Euler Taveira <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Amit Kapila @ 2022-08-11 11:04 UTC (permalink / raw)
To: Euler Taveira <[email protected]>; +Cc: Melih Mutlu <[email protected]>; pgsql-hackers
On Thu, Aug 11, 2022 at 7:34 AM Euler Taveira <[email protected]> wrote:
>
> On Wed, Aug 10, 2022, at 12:03 PM, Melih Mutlu wrote:
>
> I see that logical replication subscriptions have an option to enable binary [1].
> When it's enabled, subscription requests publisher to send data in binary format.
> But this is only the case for apply phase. In tablesync, tables are still copied as text.
>
> This option could have been included in the commit 9de77b54531; it wasn't.
> Maybe it wasn't considered because the initial table synchronization can be a
> separate step in your logical replication setup idk. I agree that the binary
> option should be available for the initial table synchronization.
>
> To copy tables, COPY command is used and that command supports copying in binary. So it seemed to me possible to copy in binary for tablesync too.
> I'm not sure if there is a reason to always copy tables in text format. But I couldn't see why not to do it in binary if it's enabled.
>
> The reason to use text format is that it is error prone. There are restrictions
> while using the binary format. For example, if your schema has different data
> types for a certain column, the copy will fail.
>
Won't such restrictions hold true even during replication?
--
With Regards,
Amit Kapila.
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-11 13:26 Euler Taveira <[email protected]>
parent: Amit Kapila <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Euler Taveira @ 2022-08-11 13:26 UTC (permalink / raw)
To: Amit Kapila <[email protected]>; +Cc: Melih Mutlu <[email protected]>; pgsql-hackers
On Thu, Aug 11, 2022, at 8:04 AM, Amit Kapila wrote:
> On Thu, Aug 11, 2022 at 7:34 AM Euler Taveira <[email protected]> wrote:
> >
> > The reason to use text format is that it is error prone. There are restrictions
> > while using the binary format. For example, if your schema has different data
> > types for a certain column, the copy will fail.
> >
>
> Won't such restrictions hold true even during replication?
I expect that the COPY code matches the proto.c code. The point is that table
sync is decoupled from the logical replication. Hence, we should emphasize in
the documentation that the restrictions *also* apply to the initial table
synchronization.
--
Euler Taveira
EDB https://www.enterprisedb.com/
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-11 13:46 Melih Mutlu <[email protected]>
parent: Euler Taveira <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Melih Mutlu @ 2022-08-11 13:46 UTC (permalink / raw)
To: Euler Taveira <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-hackers
Euler Taveira <[email protected]>, 11 Ağu 2022 Per, 16:27 tarihinde şunu
yazdı:
> On Thu, Aug 11, 2022, at 8:04 AM, Amit Kapila wrote:
>
> On Thu, Aug 11, 2022 at 7:34 AM Euler Taveira <[email protected]> wrote:
> >
> > The reason to use text format is that it is error prone. There are
> restrictions
> > while using the binary format. For example, if your schema has different
> data
> > types for a certain column, the copy will fail.
> >
>
> Won't such restrictions hold true even during replication?
>
> I expect that the COPY code matches the proto.c code. The point is that
> table
> sync is decoupled from the logical replication. Hence, we should emphasize
> in
> the documentation that the restrictions *also* apply to the initial table
> synchronization.
>
If such restrictions are already the case for replication phase after
initial table sync, then it shouldn't prevent us from enabling binary
option for table sync. Right?
But yes, it needs to be stated somewhere.
Euler Taveira <[email protected]>, 11 Ağu 2022 Per, 05:03 tarihinde şunu
yazdı
> I have a few points about your implementation.
>
> * Are we considering to support prior Postgres versions too? These releases
> support binary mode but it could be an unexpected behavior (initial sync
> in
> binary mode) for a publisher using 14 or 15 and a subscriber using 16.
> IMO
> you should only allow it for publisher on 16 or later.
>
How is any issue that might occur due to version mismatch being handled
right now in repliaction after table sync?
What I understand from the documentation is if replication can fail due to
using different pg versions, it just fails. So binary option cannot be
used. [1]
Do you think that this is more serious for table sync and we need to
restrict binary option with different publisher and subscriber versions?
But not for replication?
* Docs should say that the binary option also applies to initial table
> synchronization and possibly emphasize some of the restrictions.
> * Tests. Are the current tests enough? 014_binary.pl.
>
You're right on both points. I just wanted to know your opinions on this
first. Then the patch will need some tests and proper documentation.
[1] https://www.postgresql.org/docs/15/sql-createsubscription.html
<https://www.postgresql.org/docs/15/sql-createsubscription.html;
Best,
Melih
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-11 17:15 Euler Taveira <[email protected]>
parent: Melih Mutlu <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Euler Taveira @ 2022-08-11 17:15 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-hackers
On Thu, Aug 11, 2022, at 10:46 AM, Melih Mutlu wrote:
> If such restrictions are already the case for replication phase after initial table sync, then it shouldn't prevent us from enabling binary option for table sync. Right?
I didn't carefully examine the COPY code but I won't expect significant
differences (related to text vs binary mode) from the logical replication
protocol. After inspecting the git history, I took my argument back after
checking the commit 670c0a1d474. The initial commit 9de77b54531 imposes some
restrictions (user-defined arrays and composite types) as mentioned in the
commit message but it was removed in 670c0a1d474. My main concern is to break a
scenario that was previously working (14 -> 15) but after a subscriber upgrade
it won't (14 -> 16). I would say that you should test some scenarios:
014_binary.pl and also custom data types, same column with different data
types, etc.
> How is any issue that might occur due to version mismatch being handled right now in repliaction after table sync?
> What I understand from the documentation is if replication can fail due to using different pg versions, it just fails. So binary option cannot be used. [1]
> Do you think that this is more serious for table sync and we need to restrict binary option with different publisher and subscriber versions? But not for replication?
It is a conservative argument. If we didn't allow a publisher to run COPY in
binary mode while using previous Postgres versions, we know that it works. (At
least there aren't bug reports for logical replication using binary option.)
Since one of the main use cases for logical replication is migration, I'm
concerned that it may not work (even if the binary option defaults to false,
someone can decide to use it for performance reasons).
I did a quick test and the failure while using binary mode is not clear. Since
you are modifying this code, you could probably provide additional patch(es) to
make it clear that there is an error (due to some documented restriction).
--
Euler Taveira
EDB https://www.enterprisedb.com/
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-08-15 17:03 Melih Mutlu <[email protected]>
parent: Euler Taveira <[email protected]>
0 siblings, 2 replies; 18+ messages in thread
From: Melih Mutlu @ 2022-08-15 17:03 UTC (permalink / raw)
To: Euler Taveira <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-hackers
Euler Taveira <[email protected]>, 11 Ağu 2022 Per, 20:16 tarihinde şunu
yazdı:
> My main concern is to break a scenario that was previously working (14 ->
> 15) but after a subscriber upgrade
> it won't (14 -> 16).
>
Fair concern. Some cases that might break the logical replication with
version upgrade would be:
1- Usage of different binary formats between publisher and subscriber. As
stated here [1], binary format has been changed after v7.4.
But I don't think this would be a concern, since we wouldn't even have
logical replication with 7.4 and earlier versions.
2- Lack (or mismatch) of binary send/receive functions for custom data
types would cause failures. This case can already cause failures with
current logical replication, regardless of binary copy. Stated here [2].
3- Copying in binary format would work with the same schemas. Currently,
logical replication does not require the exact same schemas in publisher
and subscriber.
This is an additional restriction that comes with the COPY command.
If a logical replication has been set up with different schemas and
subscription is created with the binary option, then yes this would break
things.
This restriction can be clearly stated and wouldn't be unexpected though.
I'm also okay with allowing binary copy only for v16 or later, if you think
it would be safer and no one disagrees with that.
What are your thoughts?
I would say that you should test some scenarios:
> 014_binary.pl and also custom data types, same column with different data
> types, etc.
>
I added scenarios in two tests to test binary copy:
014_binary.pl: This was already testing subscriptions with binary option
enabled. I added an extra step to insert initial data before creating the
subscription.
So that we can test initial table sync with binary copy.
002_types.pl: This file was already testing more complex data types. I
added an extra subscriber node to create a subscription with binary option
enabled.
This way, it now tests binary copy with different custom types.
Do you think these would be enough in terms of testing?
Attached patch also includes some additions to the doc along with the
tests.
Thanks,
Melih
[1] https://www.postgresql.org/docs/devel/sql-copy.html
[2] https://www.postgresql.org/docs/devel/sql-createsubscription.html
Attachments:
[application/octet-stream] v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch (22.1K, ../../CAGPVpCRjdFHa71Qqd4Xj22UTa5=-NBUPgtT==X0w7TfWUZrpUg@mail.gmail.com/3-v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From 0bba087f29019e0ca0cc6f8996db4f21533b0767 Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH] Allow logical replication to copy table in binary
If binary option is enabled in a subscription, copy tables in binary
format during table synchronization.
This commit modifies tests 014_binary.pl and 002_types.pl to
test initial table synchronization step in binary format
---
doc/src/sgml/ref/create_subscription.sgml | 8 +-
src/backend/replication/logical/tablesync.c | 11 +-
src/test/subscription/t/002_types.pl | 263 ++++++++++++++++----
src/test/subscription/t/014_binary.pl | 28 ++-
4 files changed, 248 insertions(+), 62 deletions(-)
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 7390c715bc..9a576a2b95 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -183,14 +183,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<term><literal>binary</literal> (<type>boolean</type>)</term>
<listitem>
<para>
- Specifies whether the subscription will request the publisher to
- send the data in binary format (as opposed to text).
+ Specifies whether the subscription will copy the initial data to
+ synchronize relations in binary format and also request the publisher
+ to send the data in binary format (as opposed to text).
The default is <literal>false</literal>.
Even when this option is enabled, only data types having
binary send and receive functions will be transferred in binary.
</para>
<para>
+ Since logical replication uses <command>COPY</command> command,
+ it inherits restrictions of <command>COPY</command> command for
+ binary format.(See <xref linkend="sql-copy"/>.)
When doing cross-version replication, it could be that the
publisher has a binary send function for some data type, but the
subscriber lacks a binary receive function for that type. In
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 6a01ffd273..a3b128f6fb 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -101,6 +101,7 @@
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
+#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
@@ -1032,6 +1033,7 @@ copy_table(Relation rel)
CopyFromState cstate;
List *attnamelist;
ParseState *pstate;
+ List *options = NIL;
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
@@ -1110,6 +1112,13 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ if (MySubscription->binary)
+ {
+ appendStringInfoString(&cmd, " WITH (FORMAT binary)");
+ options = lappend(options, makeDefElem("format", (Node *) makeString("binary"), -1));
+ }
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1126,7 +1135,7 @@ copy_table(Relation rel)
NULL, false, false);
attnamelist = make_copy_attnamelist(relmapentry);
- cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
+ cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
/* Do the copy */
(void) CopyFrom(cstate);
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index d6c6f49327..4386033962 100644
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -19,6 +19,11 @@ my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;
+# Create binary subscriber node
+my $node_subscriber_binary = PostgreSQL::Test::Cluster->new('subscriber_binary');
+$node_subscriber_binary->init(allows_streaming => 'logical');
+$node_subscriber_binary->start;
+
# Create some preexisting content on publisher
my $ddl = qq(
CREATE EXTENSION hstore WITH SCHEMA public;
@@ -104,6 +109,100 @@ my $ddl = qq(
# Setup structure on both nodes
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+# Insert initial test data
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ -- test_tbl_one_array_col
+ INSERT INTO tst_one_array (a, b) VALUES
+ (1, '{1, 2, 3}'),
+ (2, '{2, 3, 1}');
+
+ -- test_tbl_arrays
+ INSERT INTO tst_arrays (a, b, c, d) VALUES
+ ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
+ ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}');
+
+ -- test_tbl_single_enum
+ INSERT INTO tst_one_enum (a, b) VALUES
+ (1, 'a'),
+ (2, 'b');
+
+ -- test_tbl_enums
+ INSERT INTO tst_enums (a, b) VALUES
+ ('a', '{b, c}'),
+ ('b', '{c, a}');
+
+ -- test_tbl_single_composites
+ INSERT INTO tst_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composites
+ INSERT INTO tst_comps (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]);
+
+ -- test_tbl_composite_with_enums
+ INSERT INTO tst_comp_enum (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composite_with_enums_array
+ INSERT INTO tst_comp_enum_array (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]);
+
+ -- test_tbl_composite_with_single_enums_array_in_composite
+ INSERT INTO tst_comp_one_enum_array (a, b) VALUES
+ (1, ROW(1.0, '{a, b, c}', 1)),
+ (2, ROW(2.0, '{a, b, c}', 2));
+
+ -- test_tbl_composite_with_enums_array_in_composite
+ INSERT INTO tst_comp_enum_what (a, b) VALUES
+ (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
+ (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]);
+
+ -- test_tbl_mixed_composites
+ INSERT INTO tst_comp_mix_array (a, b) VALUES
+ (ROW(
+ ROW(1,'a',1),
+ ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
+ 'a',
+ '{a,b,NULL,c}'),
+ ARRAY[
+ ROW(
+ ROW(1,'a',1),
+ ARRAY[
+ ROW(1,'a',1)::tst_comp_basic_t,
+ ROW(2,'b',2)::tst_comp_basic_t,
+ NULL
+ ],
+ 'a',
+ '{a,b,c}'
+ )::tst_comp_mix_t
+ ]
+ );
+
+ -- test_tbl_range
+ INSERT INTO tst_range (a, b) VALUES
+ (1, '[1, 10]'),
+ (2, '[2, 20]');
+
+ -- test_tbl_range_array
+ INSERT INTO tst_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+
+ -- tst_hstore
+ INSERT INTO tst_hstore (a, b) VALUES
+ (1, '"a"=>"1"'),
+ (2, '"zzz"=>"foo"');
+
+ -- tst_dom_constr
+ INSERT INTO tst_dom_constr VALUES (10);
+));
# Setup logical replication
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
@@ -113,89 +212,126 @@ $node_publisher->safe_psql('postgres',
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot)"
);
+$node_subscriber_binary->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_binary CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_binary_slot, binary = true)"
+);
# Wait for initial sync to finish as well
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+$node_subscriber_binary->wait_for_subscription_sync($node_publisher, 'tap_sub_binary');
+
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b FROM tst_one_array ORDER BY a;
+ SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+ SELECT a, b FROM tst_one_enum ORDER BY a;
+ SELECT a, b FROM tst_enums ORDER BY a;
+ SELECT a, b FROM tst_one_comp ORDER BY a;
+ SELECT a, b FROM tst_comps ORDER BY a;
+ SELECT a, b FROM tst_comp_enum ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+ SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+ SELECT a, b FROM tst_range ORDER BY a;
+ SELECT a, b, c FROM tst_range_array ORDER BY a;
+ SELECT a, b FROM tst_hstore ORDER BY a;
+);
+
+# Check the synced data on subscribers
+my $result = $node_subscriber->safe_psql('postgres', $sync_check);
+my $result_binary = $node_subscriber_binary->safe_psql('postgres', $sync_check);
+
+my $sync_result = '1|{1,2,3}
+2|{2,3,1}
+{1,2,3}|{a,b,c}|{1.1,2.2,3.3}|{"1 day","2 days","3 days"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+1|a
+2|b
+a|{b,c}
+b|{c,a}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,"{a,b,c}",1)
+2|(2,"{a,b,c}",2)
+(1,"{a,b,c}",1)|{"(1,\"{a,b,c}\",1)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+1|[1,11)
+2|[2,21)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+1|"a"=>"1"
+2|"zzz"=>"foo"';
+
+is( $result, $sync_result, 'check initial sync on subscriber');
+is( $result_binary, $sync_result, 'check initial sync on subscriber in binary');
# Insert initial test data
$node_publisher->safe_psql(
'postgres', qq(
-- test_tbl_one_array_col
INSERT INTO tst_one_array (a, b) VALUES
- (1, '{1, 2, 3}'),
- (2, '{2, 3, 1}'),
(3, '{3, 2, 1}'),
(4, '{4, 3, 2}'),
(5, '{5, NULL, 3}');
-- test_tbl_arrays
INSERT INTO tst_arrays (a, b, c, d) VALUES
- ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
- ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}'),
('{3, 1, 2}', '{"c", "a", "b"}', '{3.3, 1.1, 2.2}', '{"3 years", "1 year", "2 years"}'),
('{4, 1, 2}', '{"d", "a", "b"}', '{4.4, 1.1, 2.2}', '{"4 years", "1 year", "2 years"}'),
('{5, NULL, NULL}', '{"e", NULL, "b"}', '{5.5, 1.1, NULL}', '{"5 years", NULL, NULL}');
-- test_tbl_single_enum
INSERT INTO tst_one_enum (a, b) VALUES
- (1, 'a'),
- (2, 'b'),
(3, 'c'),
(4, 'd'),
(5, NULL);
-- test_tbl_enums
INSERT INTO tst_enums (a, b) VALUES
- ('a', '{b, c}'),
- ('b', '{c, a}'),
('c', '{b, a}'),
('d', '{c, b}'),
('e', '{d, NULL}');
-- test_tbl_single_composites
INSERT INTO tst_one_comp (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, NULL, 5));
-- test_tbl_composites
INSERT INTO tst_comps (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_basic_t]),
(ROW(4.0, 'd', 4), ARRAY[ROW(4, 'd', 3)::tst_comp_basic_t]),
(ROW(5.0, 'e', NULL), ARRAY[NULL, ROW(5, NULL, 5)::tst_comp_basic_t]);
-- test_tbl_composite_with_enums
INSERT INTO tst_comp_enum (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, 'e', NULL));
-- test_tbl_composite_with_enums_array
INSERT INTO tst_comp_enum_array (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_enum_t]),
(ROW(4.0, 'd', 3), ARRAY[ROW(3, 'd', 3)::tst_comp_enum_t]),
(ROW(5.0, 'e', 3), ARRAY[ROW(3, 'e', 3)::tst_comp_enum_t, NULL]);
-- test_tbl_composite_with_single_enums_array_in_composite
INSERT INTO tst_comp_one_enum_array (a, b) VALUES
- (1, ROW(1.0, '{a, b, c}', 1)),
- (2, ROW(2.0, '{a, b, c}', 2)),
(3, ROW(3.0, '{a, b, c}', 3)),
(4, ROW(4.0, '{c, b, d}', 4)),
(5, ROW(5.0, '{NULL, e, NULL}', 5));
-- test_tbl_composite_with_enums_array_in_composite
INSERT INTO tst_comp_enum_what (a, b) VALUES
- (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
- (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]),
(ROW(3.0, '{c, a, b}', 1), ARRAY[ROW(3, '{c, a, b}', 1)::tst_comp_enum_array_t]),
(ROW(4.0, '{c, b, d}', 4), ARRAY[ROW(4, '{c, b, d}', 4)::tst_comp_enum_array_t]),
(ROW(5.0, '{c, NULL, b}', NULL), ARRAY[ROW(5, '{c, e, b}', 1)::tst_comp_enum_array_t]);
@@ -203,10 +339,10 @@ $node_publisher->safe_psql(
-- test_tbl_mixed_composites
INSERT INTO tst_comp_mix_array (a, b) VALUES
(ROW(
- ROW(1,'a',1),
- ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
- 'a',
- '{a,b,NULL,c}'),
+ ROW(2,'b',2),
+ ARRAY[ROW(2,'b',2)::tst_comp_basic_t, ROW(3,'c',3)::tst_comp_basic_t],
+ 'b',
+ '{b,c,NULL,d}'),
ARRAY[
ROW(
ROW(1,'a',1),
@@ -223,36 +359,29 @@ $node_publisher->safe_psql(
-- test_tbl_range
INSERT INTO tst_range (a, b) VALUES
- (1, '[1, 10]'),
- (2, '[2, 20]'),
(3, '[3, 30]'),
(4, '[4, 40]'),
(5, '[5, 50]');
-- test_tbl_range_array
INSERT INTO tst_range_array (a, b, c) VALUES
- (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
- (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}'),
(3, tstzrange('Fri Aug 01 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[3,4]"}'),
(4, tstzrange('Thu Jul 31 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[4,5]", NULL, "[40,50]"}'),
(5, NULL, NULL);
-- tst_hstore
INSERT INTO tst_hstore (a, b) VALUES
- (1, '"a"=>"1"'),
- (2, '"zzz"=>"foo"'),
(3, '"123"=>"321"'),
(4, '"yellow horse"=>"moaned"');
-- tst_dom_constr
- INSERT INTO tst_dom_constr VALUES (10);
+ INSERT INTO tst_dom_constr VALUES (11);
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-my $result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $initial_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -268,9 +397,13 @@ my $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $initial_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $initial_check);
-is( $result, '1|{1,2,3}
+my $initial_result = '1|{1,2,3}
2|{2,3,1}
3|{3,2,1}
4|{4,3,2}
@@ -321,6 +454,7 @@ e|{d,NULL}
(4,"{c,b,d}",4)|{"(4,\"{c,b,d}\",4)"}
(5,"{c,NULL,b}",)|{"(5,\"{c,e,b}\",1)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[1,11)
2|[2,21)
3|[3,31)
@@ -334,8 +468,10 @@ e|{d,NULL}
1|"a"=>"1"
2|"zzz"=>"foo"
3|"123"=>"321"
-4|"yellow horse"=>"moaned"',
- 'check replicated inserts on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $initial_result, 'check replicated inserts on subscriber');
+is( $result_binary, $initial_result, 'check replicated inserts on subscriber in binary');
# Run batch of updates
$node_publisher->safe_psql(
@@ -370,10 +506,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $update_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -389,9 +524,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
-is( $result, '1|{4,5,6}
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $update_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $update_check);
+
+my $update_result = '1|{4,5,6}
2|{2,3,1}
3|{3,2,1}
4|{4,5,6,1}
@@ -442,6 +581,7 @@ e|{e,d}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")",NULL}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[100,1001)
2|[2,21)
3|[3,31)
@@ -455,8 +595,10 @@ e|{e,d}
1|"updated"=>"value"
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated updates on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $update_result, 'check replicated updates on subscriber');
+is( $result_binary, $update_result, 'check replicated updates on subscriber in binary');
# Run batch of deletes
$node_publisher->safe_psql(
@@ -490,10 +632,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $delete_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -509,9 +650,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
-is( $result, '3|{3,2,1}
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $delete_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $delete_check);
+
+my $delete_result = '3|{3,2,1}
4|{4,5,6,1}
5|{4,5,6,1}
{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
@@ -540,26 +685,36 @@ e|{e,d}
(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated deletes on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $delete_result, 'check replicated deletes on subscriber');
+is( $result_binary, $delete_result, 'check replicated deletes on subscriber in binary');
# Test a domain with a constraint backed by a SQL-language function,
# which needs an active snapshot in order to operate.
$node_publisher->safe_psql('postgres',
- "INSERT INTO tst_dom_constr VALUES (11)");
+ "INSERT INTO tst_dom_constr VALUES (12)");
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
$result =
$node_subscriber->safe_psql('postgres',
"SELECT sum(a) FROM tst_dom_constr");
-is($result, '21', 'sql-function constraint on domain');
+is($result, '33', 'sql-function constraint on domain');
+
+$result_binary =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT sum(a) FROM tst_dom_constr");
+is($result_binary, '33', 'sql-function constraint on domain');
$node_subscriber->stop('fast');
+$node_subscriber_binary->stop('fast');
$node_publisher->stop('fast');
done_testing();
diff --git a/src/test/subscription/t/014_binary.pl b/src/test/subscription/t/014_binary.pl
index 8d8b35721f..395efee1c7 100644
--- a/src/test/subscription/t/014_binary.pl
+++ b/src/test/subscription/t/014_binary.pl
@@ -36,6 +36,16 @@ my $ddl = qq(
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+# Insert some content and make sure it's synced to subscriber
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ INSERT INTO public.test_arrays (a, b, c) VALUES
+ ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}');
+
+ INSERT INTO public.test_numerical (a, b, c, d) VALUES
+ (1, 1.2, 1.3, 10);
+ ));
+
# Configure logical replication
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tpub FOR ALL TABLES");
@@ -48,27 +58,35 @@ $node_subscriber->safe_psql('postgres',
# Ensure nodes are in sync with each other
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
+
+is( $result, '1|1.2|1.3|10
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}', 'check syned data on subscriber');
+
# Insert some content and make sure it's replicated across
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_arrays (a, b, c) VALUES
- ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}'),
('{3,1,2}', '{1.3, 1.1, 1.2}', '{"three", "one", "two"}');
INSERT INTO public.test_numerical (a, b, c, d) VALUES
- (1, 1.2, 1.3, 10),
(2, 2.2, 2.3, 20),
(3, 3.2, 3.3, 30);
));
$node_publisher->wait_for_catchup('tsub');
-my $result = $node_subscriber->safe_psql('postgres',
- "SELECT a, b, c, d FROM test_numerical ORDER BY a");
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
is( $result, '1|1.2|1.3|10
2|2.2|2.3|20
-3|3.2|3.3|30', 'check replicated data on subscriber');
+3|3.2|3.3|30
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}
+{3,1,2}|{1.3,1.1,1.2}|{three,one,two}', 'check replicated data on subscriber');
# Test updates as well
$node_publisher->safe_psql(
--
2.25.1
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-09-01 22:25 Andres Freund <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 1 reply; 18+ messages in thread
From: Andres Freund @ 2022-09-01 22:25 UTC (permalink / raw)
To: Melih Mutlu <[email protected]>; +Cc: pgsql-hackers
Hi,
On 2022-08-10 18:03:56 +0300, Melih Mutlu wrote:
> To copy tables, COPY command is used and that command supports copying in
> binary. So it seemed to me possible to copy in binary for tablesync too.
> I'm not sure if there is a reason to always copy tables in text format.
It'd be good to collect some performance numbers justifying this. I'd expect
decent gains if there's e.g. a bytea or timestamptz column involved.
Greetings,
Andres Freund
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-09-07 11:51 Melih Mutlu <[email protected]>
parent: Andres Freund <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Melih Mutlu @ 2022-09-07 11:51 UTC (permalink / raw)
To: Andres Freund <[email protected]>; +Cc: pgsql-hackers
Hello,
Andres Freund <[email protected]>, 2 Eyl 2022 Cum, 01:25 tarihinde şunu
yazdı:
> It'd be good to collect some performance numbers justifying this. I'd
> expect
> decent gains if there's e.g. a bytea or timestamptz column involved.
Experimented the binary copy with a quick setup.
- Created a "temp" table with bytea and timestamptz columns
> postgres=# \d temp
> Table "public.temp"
> Column | Type | Collation | Nullable | Default
> --------+--------------------------+-----------+----------+---------
> i | integer | | |
> b | bytea | | |
> t | timestamp with time zone | | |
>
- Loaded with ~1GB data
> postgres=# SELECT pg_size_pretty( pg_total_relation_size('temp') );
> pg_size_pretty
> ----------------
> 1137 MB
> (1 row)
- Created a publication with only this "temp" table.
- Created a subscription with binary enabled on instances from master
branch and this patch.
- Timed the tablesync process by calling the following procedure:
> CREATE OR REPLACE PROCEDURE wait_for_rep() LANGUAGE plpgsql AS $$BEGIN
> WHILE (SELECT count(*) != 0 FROM pg_subscription_rel WHERE srsubstate <>
> 'r') LOOP COMMIT; END LOOP; END; $$;
Hera are averaged results of multiple consecutive runs from both master
branch and the patch:
master (binary enabled but no binary copy): 20007.7948 ms
the patch (allows binary copy): 8874,869 ms
Seems like a good improvement.
What are your thoughts on this patch?
Best,
Melih
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-09-14 16:50 Melih Mutlu <[email protected]>
parent: Melih Mutlu <[email protected]>
0 siblings, 0 replies; 18+ messages in thread
From: Melih Mutlu @ 2022-09-14 16:50 UTC (permalink / raw)
To: pgsql-hackers; +Cc: Amit Kapila <[email protected]>; Andres Freund <[email protected]>; Euler Taveira <[email protected]>
Hi hackers,
I just wanted to gently ping to hear what you all think about this patch.
Appreciate any feedback/thougths.
Thanks,
Melih
^ permalink raw reply [nested|flat] 18+ messages in thread
* RE: Allow logical replication to copy tables in binary format
@ 2022-09-16 13:51 [email protected] <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 1 reply; 18+ messages in thread
From: [email protected] @ 2022-09-16 13:51 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; Euler Taveira <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-hackers
On Tuesday, August 16, 2022 2:04 AM Melih Mutlu <[email protected]> wrote:
> Attached patch also includes some additions to the doc along with the tests.
Hi, thank you for updating the patch. Minor review comments for the v2.
(1) whitespace issues
Please fix below whitespace errors.
$ git apply v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch
v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch:39: trailing whitespace.
binary format.(See <xref linkend="sql-copy"/>.)
v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch:120: trailing whitespace.
v2-0001-Allow-logical-replication-to-copy-table-in-binary.patch:460: trailing whitespace.
);
warning: 3 lines add whitespace errors.
(2) Suggestion to update another general description about the subscription
Kindly have a look at doc/src/sgml/logical-replication.sgml.
"The data types of the columns do not need to match,
as long as the text representation of the data can be converted to the target type.
For example, you can replicate from a column of type integer to a column of type bigint."
With the patch, I think we have an impact about those descriptions
since enabling the binary option for a subscription and executing the
initial synchronization requires the same data types for binary format.
I suggest that we update those descriptions as well.
(3) shouldn't test that we fail expectedly with binary copy for different types ?
How about having a test that we correctly fail with different data types
between the publisher and the subscriber, for instance ?
(4) Error message of the binary format copy
I've gotten below message from data type contradiction (between integer and bigint).
Probably, this is unclear for the users to understand the direct cause
and needs to be adjusted ?
This might be a similar comment Euler mentioned in [1].
2022-09-16 11:54:54.835 UTC [4570] ERROR: insufficient data left in message
2022-09-16 11:54:54.835 UTC [4570] CONTEXT: COPY tab, line 1, column id
(5) Minor adjustment of the test comment in 002_types.pl.
+is( $result, $sync_result, 'check initial sync on subscriber');
+is( $result_binary, $sync_result, 'check initial sync on subscriber in binary');
# Insert initial test data
There are two same comments which say "Insert initial test data" in this file.
We need to update them, one for the initial table sync and
the other for the application of changes.
[1] - https://www.postgresql.org/message-id/f1d58324-8df4-4bb5-a546-8c741c2e6fa8%40www.fastmail.com
Best Regards,
Takamichi Osumi
^ permalink raw reply [nested|flat] 18+ messages in thread
* RE: Allow logical replication to copy tables in binary format
@ 2022-09-22 03:22 [email protected] <[email protected]>
parent: Melih Mutlu <[email protected]>
1 sibling, 0 replies; 18+ messages in thread
From: [email protected] @ 2022-09-22 03:22 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; Euler Taveira <[email protected]>; +Cc: Amit Kapila <[email protected]>; pgsql-hackers
Hi
Few more minor comments.
On Tuesday, August 16, 2022 2:04 AM Melih Mutlu <[email protected]> wrote:
>
>
> My main concern is to break a scenario that was previously working (14
> -> 15) but after a subscriber upgrade
> it won't (14 -> 16).
>
> Fair concern. Some cases that might break the logical replication with version
> upgrade would be:
...
> 3- Copying in binary format would work with the same schemas. Currently,
> logical replication does not require the exact same schemas in publisher and
> subscriber.
> This is an additional restriction that comes with the COPY command.
>
> If a logical replication has been set up with different schemas and subscription
> is created with the binary option, then yes this would break things.
> This restriction can be clearly stated and wouldn't be unexpected though.
>
> I'm also okay with allowing binary copy only for v16 or later, if you think it would
> be safer and no one disagrees with that.
> What are your thoughts?
I agree with the direction to support binary copy for v16 and later.
IIUC, the binary format replication with different data types fails even during apply phase on HEAD.
I thought that means, the upgrade concern only applies to a scenario that the user executes
only initial table synchronizations between the publisher and subscriber
and doesn't replicate any data at apply phase after that. I would say
this isn't a valid scenario and your proposal makes sense.
Best Regards,
Takamichi Osumi
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-10-03 11:50 Melih Mutlu <[email protected]>
parent: [email protected] <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: Melih Mutlu @ 2022-10-03 11:50 UTC (permalink / raw)
To: [email protected] <[email protected]>; +Cc: Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
Hi Takamichi,
Thanks for your reviews.
I addressed your reviews, please find the attached patch.
[email protected] <[email protected]>, 16 Eyl 2022 Cum,
16:51 tarihinde şunu yazdı:
> (1) whitespace issues
>
Fixed
(2) Suggestion to update another general description about the subscription
>
> Kindly have a look at doc/src/sgml/logical-replication.sgml.
>
> "The data types of the columns do not need to match,
> as long as the text representation of the data can be converted to the
> target type.
> For example, you can replicate from a column of type integer to a column
> of type bigint."
>
> With the patch, I think we have an impact about those descriptions
> since enabling the binary option for a subscription and executing the
> initial synchronization requires the same data types for binary format.
>
> I suggest that we update those descriptions as well.
>
You're right, this needs to be stated in the docs. Modified descriptions
accordingly.
> (3) shouldn't test that we fail expectedly with binary copy for different
> types ?
>
> How about having a test that we correctly fail with different data types
> between the publisher and the subscriber, for instance ?
>
Modified 002_types.pl test such that it now tests the replication between
different data types.
It's expected to fail if the binary is enabled, and succeed if not.
> (4) Error message of the binary format copy
>
> I've gotten below message from data type contradiction (between integer
> and bigint).
> Probably, this is unclear for the users to understand the direct cause
> and needs to be adjusted ?
> This might be a similar comment Euler mentioned in [1].
>
> 2022-09-16 11:54:54.835 UTC [4570] ERROR: insufficient data left in
> message
> 2022-09-16 11:54:54.835 UTC [4570] CONTEXT: COPY tab, line 1, column id
>
It's already unclear for users to understand what's the issue if they're
copying data between different column types via the COPY command.
This issue comes from COPY, and logical replication just utilizes COPY.
I don't think it would make sense to adjust an error message from a
functionality which logical replication only uses and has no direct impact
on.
It might be better to do this in a separate patch. What do you think?
> (5) Minor adjustment of the test comment in 002_types.pl.
>
> +is( $result, $sync_result, 'check initial sync on subscriber');
> +is( $result_binary, $sync_result, 'check initial sync on subscriber in
> binary');
>
> # Insert initial test data
>
> There are two same comments which say "Insert initial test data" in this
> file.
> We need to update them, one for the initial table sync and
> the other for the application of changes.
>
Fixed.
I agree with the direction to support binary copy for v16 and later.
>
> IIUC, the binary format replication with different data types fails even
> during apply phase on HEAD.
> I thought that means, the upgrade concern only applies to a scenario that
> the user executes
> only initial table synchronizations between the publisher and subscriber
> and doesn't replicate any data at apply phase after that. I would say
> this isn't a valid scenario and your proposal makes sense.
>
No, logical replication in binary does not fail on apply phase if data
types are different.
The concern with upgrade (if data types are not the same) would be not
being able to create a new subscription with binary enabled or replicate
new tables added into publication.
Replication of tables from existing subscriptions would not be affected by
this change since they will already be in the apply phase, not tablesync.
Do you think this would still be an issue?
Thanks,
Melih
Attachments:
[application/octet-stream] v3-0001-Allow-logical-replication-to-copy-table-in-binary.patch (24.3K, ../../CAGPVpCR_g8c1xXfR4kkSXiSt314C0it0seRzfqgTd+oF7N_5jw@mail.gmail.com/3-v3-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From 64b730b7f60da4742f313043076c6f150671d34b Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH] Allow logical replication to copy table in binary
If binary option is enabled in a subscription, copy tables in binary
format during table synchronization.
---
doc/src/sgml/logical-replication.sgml | 12 +-
doc/src/sgml/ref/create_subscription.sgml | 8 +-
src/backend/replication/logical/tablesync.c | 11 +-
src/test/subscription/t/002_types.pl | 287 ++++++++++++++++----
src/test/subscription/t/014_binary.pl | 28 +-
5 files changed, 278 insertions(+), 68 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index e98538e540..249b386abc 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -239,11 +239,13 @@
Columns of a table are also matched by name. The order of columns in the
subscriber table does not need to match that of the publisher. The data
types of the columns do not need to match, as long as the text
- representation of the data can be converted to the target type. For
- example, you can replicate from a column of type <type>integer</type> to a
- column of type <type>bigint</type>. The target table can also have
- additional columns not provided by the published table. Any such columns
- will be filled with the default value as specified in the definition of the
+ representation of the data can be converted to the target type and binary option
+ is not enabled. For example, you can replicate from a column of type
+ <type>integer</type> to a column of type <type>bigint</type>. In binary case, it is
+ not allowed to replicate data between different types due to restrictions inherited from
+ <command>COPY</command> (See <xref linkend="sql-copy"/>).
+ The target table can also have additional columns not provided by the published table.
+ Any such columns will be filled with the default value as specified in the definition of the
target table.
</para>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index bd12e71e33..41263f1efd 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -183,14 +183,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<term><literal>binary</literal> (<type>boolean</type>)</term>
<listitem>
<para>
- Specifies whether the subscription will request the publisher to
- send the data in binary format (as opposed to text).
+ Specifies whether the subscription will copy the initial data to
+ synchronize relations in binary format and also request the publisher
+ to send the data in binary format (as opposed to text).
The default is <literal>false</literal>.
Even when this option is enabled, only data types having
binary send and receive functions will be transferred in binary.
</para>
<para>
+ Since logical replication uses <command>COPY</command> command,
+ it inherits restrictions of <command>COPY</command> command for
+ binary format (See <xref linkend="sql-copy"/>).
When doing cross-version replication, it could be that the
publisher has a binary send function for some data type, but the
subscriber lacks a binary receive function for that type. In
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index b4a7b4b7f6..f63ea3d14a 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -101,6 +101,7 @@
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
+#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
@@ -1066,6 +1067,7 @@ copy_table(Relation rel)
CopyFromState cstate;
List *attnamelist;
ParseState *pstate;
+ List *options = NIL;
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
@@ -1144,6 +1146,13 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ if (MySubscription->binary)
+ {
+ appendStringInfoString(&cmd, " WITH (FORMAT binary)");
+ options = lappend(options, makeDefElem("format", (Node *) makeString("binary"), -1));
+ }
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1160,7 +1169,7 @@ copy_table(Relation rel)
NULL, false, false);
attnamelist = make_copy_attnamelist(relmapentry);
- cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
+ cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
/* Do the copy */
(void) CopyFrom(cstate);
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index d6c6f49327..8c6be447d4 100644
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -19,6 +19,11 @@ my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;
+# Create binary subscriber node
+my $node_subscriber_binary = PostgreSQL::Test::Cluster->new('subscriber_binary');
+$node_subscriber_binary->init(allows_streaming => 'logical');
+$node_subscriber_binary->start;
+
# Create some preexisting content on publisher
my $ddl = qq(
CREATE EXTENSION hstore WITH SCHEMA public;
@@ -104,6 +109,100 @@ my $ddl = qq(
# Setup structure on both nodes
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+# Insert initial test data
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ -- test_tbl_one_array_col
+ INSERT INTO tst_one_array (a, b) VALUES
+ (1, '{1, 2, 3}'),
+ (2, '{2, 3, 1}');
+
+ -- test_tbl_arrays
+ INSERT INTO tst_arrays (a, b, c, d) VALUES
+ ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
+ ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}');
+
+ -- test_tbl_single_enum
+ INSERT INTO tst_one_enum (a, b) VALUES
+ (1, 'a'),
+ (2, 'b');
+
+ -- test_tbl_enums
+ INSERT INTO tst_enums (a, b) VALUES
+ ('a', '{b, c}'),
+ ('b', '{c, a}');
+
+ -- test_tbl_single_composites
+ INSERT INTO tst_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composites
+ INSERT INTO tst_comps (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]);
+
+ -- test_tbl_composite_with_enums
+ INSERT INTO tst_comp_enum (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composite_with_enums_array
+ INSERT INTO tst_comp_enum_array (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]);
+
+ -- test_tbl_composite_with_single_enums_array_in_composite
+ INSERT INTO tst_comp_one_enum_array (a, b) VALUES
+ (1, ROW(1.0, '{a, b, c}', 1)),
+ (2, ROW(2.0, '{a, b, c}', 2));
+
+ -- test_tbl_composite_with_enums_array_in_composite
+ INSERT INTO tst_comp_enum_what (a, b) VALUES
+ (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
+ (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]);
+
+ -- test_tbl_mixed_composites
+ INSERT INTO tst_comp_mix_array (a, b) VALUES
+ (ROW(
+ ROW(1,'a',1),
+ ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
+ 'a',
+ '{a,b,NULL,c}'),
+ ARRAY[
+ ROW(
+ ROW(1,'a',1),
+ ARRAY[
+ ROW(1,'a',1)::tst_comp_basic_t,
+ ROW(2,'b',2)::tst_comp_basic_t,
+ NULL
+ ],
+ 'a',
+ '{a,b,c}'
+ )::tst_comp_mix_t
+ ]
+ );
+
+ -- test_tbl_range
+ INSERT INTO tst_range (a, b) VALUES
+ (1, '[1, 10]'),
+ (2, '[2, 20]');
+
+ -- test_tbl_range_array
+ INSERT INTO tst_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+
+ -- tst_hstore
+ INSERT INTO tst_hstore (a, b) VALUES
+ (1, '"a"=>"1"'),
+ (2, '"zzz"=>"foo"');
+
+ -- tst_dom_constr
+ INSERT INTO tst_dom_constr VALUES (10);
+));
# Setup logical replication
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
@@ -113,89 +212,126 @@ $node_publisher->safe_psql('postgres',
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot)"
);
+$node_subscriber_binary->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_binary CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_binary_slot, binary = true)"
+);
# Wait for initial sync to finish as well
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+$node_subscriber_binary->wait_for_subscription_sync($node_publisher, 'tap_sub_binary');
-# Insert initial test data
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b FROM tst_one_array ORDER BY a;
+ SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+ SELECT a, b FROM tst_one_enum ORDER BY a;
+ SELECT a, b FROM tst_enums ORDER BY a;
+ SELECT a, b FROM tst_one_comp ORDER BY a;
+ SELECT a, b FROM tst_comps ORDER BY a;
+ SELECT a, b FROM tst_comp_enum ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+ SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+ SELECT a, b FROM tst_range ORDER BY a;
+ SELECT a, b, c FROM tst_range_array ORDER BY a;
+ SELECT a, b FROM tst_hstore ORDER BY a;
+);
+
+# Check the synced data on subscribers
+my $result = $node_subscriber->safe_psql('postgres', $sync_check);
+my $result_binary = $node_subscriber_binary->safe_psql('postgres', $sync_check);
+
+my $sync_result = '1|{1,2,3}
+2|{2,3,1}
+{1,2,3}|{a,b,c}|{1.1,2.2,3.3}|{"1 day","2 days","3 days"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+1|a
+2|b
+a|{b,c}
+b|{c,a}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,"{a,b,c}",1)
+2|(2,"{a,b,c}",2)
+(1,"{a,b,c}",1)|{"(1,\"{a,b,c}\",1)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+1|[1,11)
+2|[2,21)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+1|"a"=>"1"
+2|"zzz"=>"foo"';
+
+is( $result, $sync_result, 'check initial sync on subscriber');
+is( $result_binary, $sync_result, 'check initial sync on subscriber in binary');
+
+# Insert test data for update to apply changes
$node_publisher->safe_psql(
'postgres', qq(
-- test_tbl_one_array_col
INSERT INTO tst_one_array (a, b) VALUES
- (1, '{1, 2, 3}'),
- (2, '{2, 3, 1}'),
(3, '{3, 2, 1}'),
(4, '{4, 3, 2}'),
(5, '{5, NULL, 3}');
-- test_tbl_arrays
INSERT INTO tst_arrays (a, b, c, d) VALUES
- ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
- ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}'),
('{3, 1, 2}', '{"c", "a", "b"}', '{3.3, 1.1, 2.2}', '{"3 years", "1 year", "2 years"}'),
('{4, 1, 2}', '{"d", "a", "b"}', '{4.4, 1.1, 2.2}', '{"4 years", "1 year", "2 years"}'),
('{5, NULL, NULL}', '{"e", NULL, "b"}', '{5.5, 1.1, NULL}', '{"5 years", NULL, NULL}');
-- test_tbl_single_enum
INSERT INTO tst_one_enum (a, b) VALUES
- (1, 'a'),
- (2, 'b'),
(3, 'c'),
(4, 'd'),
(5, NULL);
-- test_tbl_enums
INSERT INTO tst_enums (a, b) VALUES
- ('a', '{b, c}'),
- ('b', '{c, a}'),
('c', '{b, a}'),
('d', '{c, b}'),
('e', '{d, NULL}');
-- test_tbl_single_composites
INSERT INTO tst_one_comp (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, NULL, 5));
-- test_tbl_composites
INSERT INTO tst_comps (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_basic_t]),
(ROW(4.0, 'd', 4), ARRAY[ROW(4, 'd', 3)::tst_comp_basic_t]),
(ROW(5.0, 'e', NULL), ARRAY[NULL, ROW(5, NULL, 5)::tst_comp_basic_t]);
-- test_tbl_composite_with_enums
INSERT INTO tst_comp_enum (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, 'e', NULL));
-- test_tbl_composite_with_enums_array
INSERT INTO tst_comp_enum_array (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_enum_t]),
(ROW(4.0, 'd', 3), ARRAY[ROW(3, 'd', 3)::tst_comp_enum_t]),
(ROW(5.0, 'e', 3), ARRAY[ROW(3, 'e', 3)::tst_comp_enum_t, NULL]);
-- test_tbl_composite_with_single_enums_array_in_composite
INSERT INTO tst_comp_one_enum_array (a, b) VALUES
- (1, ROW(1.0, '{a, b, c}', 1)),
- (2, ROW(2.0, '{a, b, c}', 2)),
(3, ROW(3.0, '{a, b, c}', 3)),
(4, ROW(4.0, '{c, b, d}', 4)),
(5, ROW(5.0, '{NULL, e, NULL}', 5));
-- test_tbl_composite_with_enums_array_in_composite
INSERT INTO tst_comp_enum_what (a, b) VALUES
- (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
- (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]),
(ROW(3.0, '{c, a, b}', 1), ARRAY[ROW(3, '{c, a, b}', 1)::tst_comp_enum_array_t]),
(ROW(4.0, '{c, b, d}', 4), ARRAY[ROW(4, '{c, b, d}', 4)::tst_comp_enum_array_t]),
(ROW(5.0, '{c, NULL, b}', NULL), ARRAY[ROW(5, '{c, e, b}', 1)::tst_comp_enum_array_t]);
@@ -203,10 +339,10 @@ $node_publisher->safe_psql(
-- test_tbl_mixed_composites
INSERT INTO tst_comp_mix_array (a, b) VALUES
(ROW(
- ROW(1,'a',1),
- ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
- 'a',
- '{a,b,NULL,c}'),
+ ROW(2,'b',2),
+ ARRAY[ROW(2,'b',2)::tst_comp_basic_t, ROW(3,'c',3)::tst_comp_basic_t],
+ 'b',
+ '{b,c,NULL,d}'),
ARRAY[
ROW(
ROW(1,'a',1),
@@ -223,36 +359,29 @@ $node_publisher->safe_psql(
-- test_tbl_range
INSERT INTO tst_range (a, b) VALUES
- (1, '[1, 10]'),
- (2, '[2, 20]'),
(3, '[3, 30]'),
(4, '[4, 40]'),
(5, '[5, 50]');
-- test_tbl_range_array
INSERT INTO tst_range_array (a, b, c) VALUES
- (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
- (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}'),
(3, tstzrange('Fri Aug 01 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[3,4]"}'),
(4, tstzrange('Thu Jul 31 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[4,5]", NULL, "[40,50]"}'),
(5, NULL, NULL);
-- tst_hstore
INSERT INTO tst_hstore (a, b) VALUES
- (1, '"a"=>"1"'),
- (2, '"zzz"=>"foo"'),
(3, '"123"=>"321"'),
(4, '"yellow horse"=>"moaned"');
-- tst_dom_constr
- INSERT INTO tst_dom_constr VALUES (10);
+ INSERT INTO tst_dom_constr VALUES (11);
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-my $result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $initial_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -268,9 +397,13 @@ my $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $initial_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $initial_check);
-is( $result, '1|{1,2,3}
+my $initial_result = '1|{1,2,3}
2|{2,3,1}
3|{3,2,1}
4|{4,3,2}
@@ -321,6 +454,7 @@ e|{d,NULL}
(4,"{c,b,d}",4)|{"(4,\"{c,b,d}\",4)"}
(5,"{c,NULL,b}",)|{"(5,\"{c,e,b}\",1)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[1,11)
2|[2,21)
3|[3,31)
@@ -334,8 +468,10 @@ e|{d,NULL}
1|"a"=>"1"
2|"zzz"=>"foo"
3|"123"=>"321"
-4|"yellow horse"=>"moaned"',
- 'check replicated inserts on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $initial_result, 'check replicated inserts on subscriber');
+is( $result_binary, $initial_result, 'check replicated inserts on subscriber in binary');
# Run batch of updates
$node_publisher->safe_psql(
@@ -370,10 +506,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $update_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -389,9 +524,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $update_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $update_check);
-is( $result, '1|{4,5,6}
+my $update_result = '1|{4,5,6}
2|{2,3,1}
3|{3,2,1}
4|{4,5,6,1}
@@ -442,6 +581,7 @@ e|{e,d}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")",NULL}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[100,1001)
2|[2,21)
3|[3,31)
@@ -455,8 +595,10 @@ e|{e,d}
1|"updated"=>"value"
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated updates on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $update_result, 'check replicated updates on subscriber');
+is( $result_binary, $update_result, 'check replicated updates on subscriber in binary');
# Run batch of deletes
$node_publisher->safe_psql(
@@ -490,10 +632,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $delete_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -509,9 +650,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
-is( $result, '3|{3,2,1}
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $delete_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $delete_check);
+
+my $delete_result = '3|{3,2,1}
4|{4,5,6,1}
5|{4,5,6,1}
{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
@@ -540,26 +685,58 @@ e|{e,d}
(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated deletes on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $delete_result, 'check replicated deletes on subscriber');
+is( $result_binary, $delete_result, 'check replicated deletes on subscriber in binary');
# Test a domain with a constraint backed by a SQL-language function,
# which needs an active snapshot in order to operate.
$node_publisher->safe_psql('postgres',
- "INSERT INTO tst_dom_constr VALUES (11)");
+ "INSERT INTO tst_dom_constr VALUES (12)");
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
$result =
$node_subscriber->safe_psql('postgres',
"SELECT sum(a) FROM tst_dom_constr");
-is($result, '21', 'sql-function constraint on domain');
+is($result, '33', 'sql-function constraint on domain');
+
+$result_binary =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT sum(a) FROM tst_dom_constr");
+is($result_binary, '33', 'sql-function constraint on domain');
+
+# Test whether subscriber fails in case of column type mismatch
+$ddl = qq(
+ CREATE TABLE tst_binary_mismatch (a int);
+ INSERT INTO tst_binary_mismatch VALUES (1);
+);
+$node_publisher->safe_psql('postgres', $ddl);
+
+$ddl = qq(
+ CREATE TABLE tst_binary_mismatch (a bigint);
+);
+$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION');
+$node_subscriber_binary->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub_binary REFRESH PUBLICATION');
+
+# Binary enabled subscription should fail
+$node_subscriber_binary->wait_for_log("ERROR: insufficient data left in message");
+
+# Binary disabled subscription should succeed
+$node_publisher->wait_for_catchup('tap_sub');
$node_subscriber->stop('fast');
+$node_subscriber_binary->stop('fast');
$node_publisher->stop('fast');
done_testing();
diff --git a/src/test/subscription/t/014_binary.pl b/src/test/subscription/t/014_binary.pl
index 8d8b35721f..395efee1c7 100644
--- a/src/test/subscription/t/014_binary.pl
+++ b/src/test/subscription/t/014_binary.pl
@@ -36,6 +36,16 @@ my $ddl = qq(
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+# Insert some content and make sure it's synced to subscriber
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ INSERT INTO public.test_arrays (a, b, c) VALUES
+ ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}');
+
+ INSERT INTO public.test_numerical (a, b, c, d) VALUES
+ (1, 1.2, 1.3, 10);
+ ));
+
# Configure logical replication
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tpub FOR ALL TABLES");
@@ -48,27 +58,35 @@ $node_subscriber->safe_psql('postgres',
# Ensure nodes are in sync with each other
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
+
+is( $result, '1|1.2|1.3|10
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}', 'check syned data on subscriber');
+
# Insert some content and make sure it's replicated across
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_arrays (a, b, c) VALUES
- ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}'),
('{3,1,2}', '{1.3, 1.1, 1.2}', '{"three", "one", "two"}');
INSERT INTO public.test_numerical (a, b, c, d) VALUES
- (1, 1.2, 1.3, 10),
(2, 2.2, 2.3, 20),
(3, 3.2, 3.3, 30);
));
$node_publisher->wait_for_catchup('tsub');
-my $result = $node_subscriber->safe_psql('postgres',
- "SELECT a, b, c, d FROM test_numerical ORDER BY a");
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
is( $result, '1|1.2|1.3|10
2|2.2|2.3|20
-3|3.2|3.3|30', 'check replicated data on subscriber');
+3|3.2|3.3|30
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}
+{3,1,2}|{1.3,1.1,1.2}|{three,one,two}', 'check replicated data on subscriber');
# Test updates as well
$node_publisher->safe_psql(
--
2.25.1
^ permalink raw reply [nested|flat] 18+ messages in thread
* RE: Allow logical replication to copy tables in binary format
@ 2022-10-12 01:36 [email protected] <[email protected]>
parent: Melih Mutlu <[email protected]>
0 siblings, 1 reply; 18+ messages in thread
From: [email protected] @ 2022-10-12 01:36 UTC (permalink / raw)
To: 'Melih Mutlu' <[email protected]>; +Cc: Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
Hi,
On Monday, October 3, 2022 8:50 PM Melih Mutlu <[email protected]> wrote:
> (4) Error message of the binary format copy
>
> I've gotten below message from data type contradiction (between
> integer and bigint).
> Probably, this is unclear for the users to understand the direct cause
> and needs to be adjusted ?
> This might be a similar comment Euler mentioned in [1].
>
> 2022-09-16 11:54:54.835 UTC [4570] ERROR: insufficient data left in
> message
> 2022-09-16 11:54:54.835 UTC [4570] CONTEXT: COPY tab, line 1,
> column id
>
> It's already unclear for users to understand what's the issue if they're copying
> data between different column types via the COPY command.
> This issue comes from COPY, and logical replication just utilizes COPY.
> I don't think it would make sense to adjust an error message from a functionality
> which logical replication only uses and has no direct impact on.
> It might be better to do this in a separate patch. What do you think?
Yes, makes sense. It should be a separate patch.
> I agree with the direction to support binary copy for v16 and later.
>
> IIUC, the binary format replication with different data types fails even
> during apply phase on HEAD.
> I thought that means, the upgrade concern only applies to a scenario
> that the user executes
> only initial table synchronizations between the publisher and subscriber
> and doesn't replicate any data at apply phase after that. I would say
> this isn't a valid scenario and your proposal makes sense.
>
> No, logical replication in binary does not fail on apply phase if data types are
> different.
With HEAD, I observe in some case we fail at apply phase because of different data types like
integer vs. bigint as written scenario in [1]. In short, I think we can slightly
adjust your documentation and make it more general so that the description applies to
both table sync phase and apply phase.
I'll suggest a below change for your sentence of logical-replication.sgml.
FROM:
In binary case, it is not allowed to replicate data between different types due to restrictions inherited from COPY.
TO:
Binary format is type specific and does not allow to replicate data between different types according to its
restrictions.
If my idea above is correct, then I feel we can remove all the fixes for create_subscription.sgml.
I'm not sure if I should pursue this perspective of the document improvement
any further after this email, since this isn't essentially because of this patch.
> The concern with upgrade (if data types are not the same) would be not being
> able to create a new subscription with binary enabled or replicate new tables
> added into publication.
> Replication of tables from existing subscriptions would not be affected by this
> change since they will already be in the apply phase, not tablesync.
> Do you think this would still be an issue?
Okay, thanks for explaining this. I understand that
the upgrade concern applies to the table sync that is executed
between text format (before the patch) and binary format (after the patch).
[1] - binary format test that we fail for different types on apply phase on HEAD
<publisher>
create table tab (id integer);
insert into tab values (100);
create publication mypub for table tab;
<subscriber>
create table tab (id bigint);
create subscription mysub connection '...' publication mypub with (copy_data = false, binary = true, disable_on_error = true);
-- wait for several seconds
<subscriber>
select srsubid, srrelid, srrelid::regclass, srsubstate, srsublsn from pg_subscription_rel; -- check the status as 'r' for the relation
select * from tab; -- confirm we don't copy the initial data on the pub
<publisher>
insert into tab values (1), (2);
-- wait for several seconds
<subscriber>
select subname, subenabled from pg_subscription; -- shows 'f' for the 2nd column because of an error
select * from tab -- no records
This error doesn't happen when we adopt 'integer' on the subscriber aligned with the publisher
and we can see the two records on the subscriber.
Best Regards,
Takamichi Osumi
^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Allow logical replication to copy tables in binary format
@ 2022-11-14 12:07 Melih Mutlu <[email protected]>
parent: [email protected] <[email protected]>
0 siblings, 0 replies; 18+ messages in thread
From: Melih Mutlu @ 2022-11-14 12:07 UTC (permalink / raw)
To: [email protected] <[email protected]>; +Cc: Euler Taveira <[email protected]>; Amit Kapila <[email protected]>; pgsql-hackers
Hello,
[email protected] <[email protected]>, 12 Eki 2022 Çar,
04:36 tarihinde şunu yazdı:
> > I agree with the direction to support binary copy for v16 and
> later.
> >
> > IIUC, the binary format replication with different data types
> fails even
> > during apply phase on HEAD.
> > I thought that means, the upgrade concern only applies to a
> scenario
> > that the user executes
> > only initial table synchronizations between the publisher and
> subscriber
> > and doesn't replicate any data at apply phase after that. I would
> say
> > this isn't a valid scenario and your proposal makes sense.
> >
> > No, logical replication in binary does not fail on apply phase if data
> types are
> > different.
> With HEAD, I observe in some case we fail at apply phase because of
> different data types like
> integer vs. bigint as written scenario in [1]. In short, I think we can
> slightly
> adjust your documentation and make it more general so that the description
> applies to
> both table sync phase and apply phase.
>
Yes, you're right. I somehow had the impression that HEAD supports
replication between different types in binary.
But as can be shown in the scenario you mentioned, it does not work.
I'll suggest a below change for your sentence of logical-replication.sgml.
> FROM:
> In binary case, it is not allowed to replicate data between different
> types due to restrictions inherited from COPY.
> TO:
> Binary format is type specific and does not allow to replicate data
> between different types according to its
> restrictions.
>
In this case, this change makes sense since this patch does actually not
introduce this issue. It already exists in HEAD too.
> If my idea above is correct, then I feel we can remove all the fixes for
> create_subscription.sgml.
> I'm not sure if I should pursue this perspective of the document
> improvement
> any further after this email, since this isn't essentially because of this
> patch.
>
I'm only keeping the following change in create_subscription.sgml to
indicate binary option copies in binary format now.
> - Specifies whether the subscription will request the publisher to
> - send the data in binary format (as opposed to text).
> + Specifies whether the subscription will copy the initial data to
> + synchronize relations in binary format and also request the
> publisher
> + to send the data in binary format too (as opposed to text).
> > The concern with upgrade (if data types are not the same) would be not
> being
> > able to create a new subscription with binary enabled or replicate new
> tables
> > added into publication.
> > Replication of tables from existing subscriptions would not be affected
> by this
> > change since they will already be in the apply phase, not tablesync.
> > Do you think this would still be an issue?
> Okay, thanks for explaining this. I understand that
> the upgrade concern applies to the table sync that is executed
> between text format (before the patch) and binary format (after the patch).
>
I was thinking apply would work with different types in binary format.
Since apply also would not work, then the scenario that I tried to explain
earlier is not a concern anymore.
Attached patch with updated version of this patch.
Thanks,
Melih
Attachments:
[application/octet-stream] v4-0001-Allow-logical-replication-to-copy-table-in-binary.patch (23.6K, ../../CAGPVpCRaxRA-SqWYsWvqWPB_CfVyasvCwC2oqg9oQgZfP0QBtQ@mail.gmail.com/3-v4-0001-Allow-logical-replication-to-copy-table-in-binary.patch)
download | inline diff:
From dc4b220e3182629f7f720e2cc202ba6f0a1b0b7f Mon Sep 17 00:00:00 2001
From: Melih Mutlu <[email protected]>
Date: Mon, 8 Aug 2022 14:14:44 +0300
Subject: [PATCH] Allow logical replication to copy table in binary
If binary option is enabled in a subscription, copy tables in binary
format during table synchronization.
---
doc/src/sgml/logical-replication.sgml | 9 +-
doc/src/sgml/ref/create_subscription.sgml | 5 +-
src/backend/replication/logical/tablesync.c | 11 +-
src/test/subscription/t/002_types.pl | 287 ++++++++++++++++----
src/test/subscription/t/014_binary.pl | 28 +-
5 files changed, 273 insertions(+), 67 deletions(-)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index f8756389a3..7ca3cb1ff0 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -241,10 +241,11 @@
types of the columns do not need to match, as long as the text
representation of the data can be converted to the target type. For
example, you can replicate from a column of type <type>integer</type> to a
- column of type <type>bigint</type>. The target table can also have
- additional columns not provided by the published table. Any such columns
- will be filled with the default value as specified in the definition of the
- target table.
+ column of type <type>bigint</type>. Replication in binary format is type
+ specific and does not allow to replicate data between different types according
+ to its restrictions. The target table can also have additional columns not provided
+ by the published table. Any such columns will be filled with the default
+ value as specified in the definition of the target table.
</para>
<sect2 id="logical-replication-subscription-slot">
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index f9a1776380..b006eab4c2 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -189,8 +189,9 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<term><literal>binary</literal> (<type>boolean</type>)</term>
<listitem>
<para>
- Specifies whether the subscription will request the publisher to
- send the data in binary format (as opposed to text).
+ Specifies whether the subscription will copy the initial data to
+ synchronize relations in binary format and also request the publisher
+ to send the data in binary format too (as opposed to text).
The default is <literal>false</literal>.
Even when this option is enabled, only data types having
binary send and receive functions will be transferred in binary.
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 94e813ac53..7ba877312d 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -101,6 +101,7 @@
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "miscadmin.h"
+#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
@@ -1066,6 +1067,7 @@ copy_table(Relation rel)
CopyFromState cstate;
List *attnamelist;
ParseState *pstate;
+ List *options = NIL;
/* Get the publisher relation info. */
fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
@@ -1144,6 +1146,13 @@ copy_table(Relation rel)
appendStringInfoString(&cmd, ") TO STDOUT");
}
+
+ if (MySubscription->binary)
+ {
+ appendStringInfoString(&cmd, " WITH (FORMAT binary)");
+ options = lappend(options, makeDefElem("format", (Node *) makeString("binary"), -1));
+ }
+
res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 0, NULL);
pfree(cmd.data);
if (res->status != WALRCV_OK_COPY_OUT)
@@ -1160,7 +1169,7 @@ copy_table(Relation rel)
NULL, false, false);
attnamelist = make_copy_attnamelist(relmapentry);
- cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, NIL);
+ cstate = BeginCopyFrom(pstate, rel, NULL, NULL, false, copy_read_data, attnamelist, options);
/* Do the copy */
(void) CopyFrom(cstate);
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index d6c6f49327..8c6be447d4 100644
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -19,6 +19,11 @@ my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init(allows_streaming => 'logical');
$node_subscriber->start;
+# Create binary subscriber node
+my $node_subscriber_binary = PostgreSQL::Test::Cluster->new('subscriber_binary');
+$node_subscriber_binary->init(allows_streaming => 'logical');
+$node_subscriber_binary->start;
+
# Create some preexisting content on publisher
my $ddl = qq(
CREATE EXTENSION hstore WITH SCHEMA public;
@@ -104,6 +109,100 @@ my $ddl = qq(
# Setup structure on both nodes
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+# Insert initial test data
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ -- test_tbl_one_array_col
+ INSERT INTO tst_one_array (a, b) VALUES
+ (1, '{1, 2, 3}'),
+ (2, '{2, 3, 1}');
+
+ -- test_tbl_arrays
+ INSERT INTO tst_arrays (a, b, c, d) VALUES
+ ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
+ ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}');
+
+ -- test_tbl_single_enum
+ INSERT INTO tst_one_enum (a, b) VALUES
+ (1, 'a'),
+ (2, 'b');
+
+ -- test_tbl_enums
+ INSERT INTO tst_enums (a, b) VALUES
+ ('a', '{b, c}'),
+ ('b', '{c, a}');
+
+ -- test_tbl_single_composites
+ INSERT INTO tst_one_comp (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composites
+ INSERT INTO tst_comps (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]);
+
+ -- test_tbl_composite_with_enums
+ INSERT INTO tst_comp_enum (a, b) VALUES
+ (1, ROW(1.0, 'a', 1)),
+ (2, ROW(2.0, 'b', 2));
+
+ -- test_tbl_composite_with_enums_array
+ INSERT INTO tst_comp_enum_array (a, b) VALUES
+ (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
+ (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]);
+
+ -- test_tbl_composite_with_single_enums_array_in_composite
+ INSERT INTO tst_comp_one_enum_array (a, b) VALUES
+ (1, ROW(1.0, '{a, b, c}', 1)),
+ (2, ROW(2.0, '{a, b, c}', 2));
+
+ -- test_tbl_composite_with_enums_array_in_composite
+ INSERT INTO tst_comp_enum_what (a, b) VALUES
+ (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
+ (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]);
+
+ -- test_tbl_mixed_composites
+ INSERT INTO tst_comp_mix_array (a, b) VALUES
+ (ROW(
+ ROW(1,'a',1),
+ ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
+ 'a',
+ '{a,b,NULL,c}'),
+ ARRAY[
+ ROW(
+ ROW(1,'a',1),
+ ARRAY[
+ ROW(1,'a',1)::tst_comp_basic_t,
+ ROW(2,'b',2)::tst_comp_basic_t,
+ NULL
+ ],
+ 'a',
+ '{a,b,c}'
+ )::tst_comp_mix_t
+ ]
+ );
+
+ -- test_tbl_range
+ INSERT INTO tst_range (a, b) VALUES
+ (1, '[1, 10]'),
+ (2, '[2, 20]');
+
+ -- test_tbl_range_array
+ INSERT INTO tst_range_array (a, b, c) VALUES
+ (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+ (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}');
+
+ -- tst_hstore
+ INSERT INTO tst_hstore (a, b) VALUES
+ (1, '"a"=>"1"'),
+ (2, '"zzz"=>"foo"');
+
+ -- tst_dom_constr
+ INSERT INTO tst_dom_constr VALUES (10);
+));
# Setup logical replication
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
@@ -113,89 +212,126 @@ $node_publisher->safe_psql('postgres',
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot)"
);
+$node_subscriber_binary->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_binary CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (slot_name = tap_sub_binary_slot, binary = true)"
+);
# Wait for initial sync to finish as well
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+$node_subscriber_binary->wait_for_subscription_sync($node_publisher, 'tap_sub_binary');
-# Insert initial test data
+my $sync_check = qq(
+ SET timezone = '+2';
+ SELECT a, b FROM tst_one_array ORDER BY a;
+ SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+ SELECT a, b FROM tst_one_enum ORDER BY a;
+ SELECT a, b FROM tst_enums ORDER BY a;
+ SELECT a, b FROM tst_one_comp ORDER BY a;
+ SELECT a, b FROM tst_comps ORDER BY a;
+ SELECT a, b FROM tst_comp_enum ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+ SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+ SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+ SELECT a, b FROM tst_range ORDER BY a;
+ SELECT a, b, c FROM tst_range_array ORDER BY a;
+ SELECT a, b FROM tst_hstore ORDER BY a;
+);
+
+# Check the synced data on subscribers
+my $result = $node_subscriber->safe_psql('postgres', $sync_check);
+my $result_binary = $node_subscriber_binary->safe_psql('postgres', $sync_check);
+
+my $sync_result = '1|{1,2,3}
+2|{2,3,1}
+{1,2,3}|{a,b,c}|{1.1,2.2,3.3}|{"1 day","2 days","3 days"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+1|a
+2|b
+a|{b,c}
+b|{c,a}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,a,1)
+2|(2,b,2)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+1|(1,"{a,b,c}",1)
+2|(2,"{a,b,c}",2)
+(1,"{a,b,c}",1)|{"(1,\"{a,b,c}\",1)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+1|[1,11)
+2|[2,21)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+1|"a"=>"1"
+2|"zzz"=>"foo"';
+
+is( $result, $sync_result, 'check initial sync on subscriber');
+is( $result_binary, $sync_result, 'check initial sync on subscriber in binary');
+
+# Insert test data for update to apply changes
$node_publisher->safe_psql(
'postgres', qq(
-- test_tbl_one_array_col
INSERT INTO tst_one_array (a, b) VALUES
- (1, '{1, 2, 3}'),
- (2, '{2, 3, 1}'),
(3, '{3, 2, 1}'),
(4, '{4, 3, 2}'),
(5, '{5, NULL, 3}');
-- test_tbl_arrays
INSERT INTO tst_arrays (a, b, c, d) VALUES
- ('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
- ('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}'),
('{3, 1, 2}', '{"c", "a", "b"}', '{3.3, 1.1, 2.2}', '{"3 years", "1 year", "2 years"}'),
('{4, 1, 2}', '{"d", "a", "b"}', '{4.4, 1.1, 2.2}', '{"4 years", "1 year", "2 years"}'),
('{5, NULL, NULL}', '{"e", NULL, "b"}', '{5.5, 1.1, NULL}', '{"5 years", NULL, NULL}');
-- test_tbl_single_enum
INSERT INTO tst_one_enum (a, b) VALUES
- (1, 'a'),
- (2, 'b'),
(3, 'c'),
(4, 'd'),
(5, NULL);
-- test_tbl_enums
INSERT INTO tst_enums (a, b) VALUES
- ('a', '{b, c}'),
- ('b', '{c, a}'),
('c', '{b, a}'),
('d', '{c, b}'),
('e', '{d, NULL}');
-- test_tbl_single_composites
INSERT INTO tst_one_comp (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, NULL, 5));
-- test_tbl_composites
INSERT INTO tst_comps (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_basic_t]),
(ROW(4.0, 'd', 4), ARRAY[ROW(4, 'd', 3)::tst_comp_basic_t]),
(ROW(5.0, 'e', NULL), ARRAY[NULL, ROW(5, NULL, 5)::tst_comp_basic_t]);
-- test_tbl_composite_with_enums
INSERT INTO tst_comp_enum (a, b) VALUES
- (1, ROW(1.0, 'a', 1)),
- (2, ROW(2.0, 'b', 2)),
(3, ROW(3.0, 'c', 3)),
(4, ROW(4.0, 'd', 4)),
(5, ROW(NULL, 'e', NULL));
-- test_tbl_composite_with_enums_array
INSERT INTO tst_comp_enum_array (a, b) VALUES
- (ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
- (ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]),
(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_enum_t]),
(ROW(4.0, 'd', 3), ARRAY[ROW(3, 'd', 3)::tst_comp_enum_t]),
(ROW(5.0, 'e', 3), ARRAY[ROW(3, 'e', 3)::tst_comp_enum_t, NULL]);
-- test_tbl_composite_with_single_enums_array_in_composite
INSERT INTO tst_comp_one_enum_array (a, b) VALUES
- (1, ROW(1.0, '{a, b, c}', 1)),
- (2, ROW(2.0, '{a, b, c}', 2)),
(3, ROW(3.0, '{a, b, c}', 3)),
(4, ROW(4.0, '{c, b, d}', 4)),
(5, ROW(5.0, '{NULL, e, NULL}', 5));
-- test_tbl_composite_with_enums_array_in_composite
INSERT INTO tst_comp_enum_what (a, b) VALUES
- (ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
- (ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]),
(ROW(3.0, '{c, a, b}', 1), ARRAY[ROW(3, '{c, a, b}', 1)::tst_comp_enum_array_t]),
(ROW(4.0, '{c, b, d}', 4), ARRAY[ROW(4, '{c, b, d}', 4)::tst_comp_enum_array_t]),
(ROW(5.0, '{c, NULL, b}', NULL), ARRAY[ROW(5, '{c, e, b}', 1)::tst_comp_enum_array_t]);
@@ -203,10 +339,10 @@ $node_publisher->safe_psql(
-- test_tbl_mixed_composites
INSERT INTO tst_comp_mix_array (a, b) VALUES
(ROW(
- ROW(1,'a',1),
- ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
- 'a',
- '{a,b,NULL,c}'),
+ ROW(2,'b',2),
+ ARRAY[ROW(2,'b',2)::tst_comp_basic_t, ROW(3,'c',3)::tst_comp_basic_t],
+ 'b',
+ '{b,c,NULL,d}'),
ARRAY[
ROW(
ROW(1,'a',1),
@@ -223,36 +359,29 @@ $node_publisher->safe_psql(
-- test_tbl_range
INSERT INTO tst_range (a, b) VALUES
- (1, '[1, 10]'),
- (2, '[2, 20]'),
(3, '[3, 30]'),
(4, '[4, 40]'),
(5, '[5, 50]');
-- test_tbl_range_array
INSERT INTO tst_range_array (a, b, c) VALUES
- (1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
- (2, tstzrange('Sat Aug 02 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}'),
(3, tstzrange('Fri Aug 01 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[3,4]"}'),
(4, tstzrange('Thu Jul 31 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[4,5]", NULL, "[40,50]"}'),
(5, NULL, NULL);
-- tst_hstore
INSERT INTO tst_hstore (a, b) VALUES
- (1, '"a"=>"1"'),
- (2, '"zzz"=>"foo"'),
(3, '"123"=>"321"'),
(4, '"yellow horse"=>"moaned"');
-- tst_dom_constr
- INSERT INTO tst_dom_constr VALUES (10);
+ INSERT INTO tst_dom_constr VALUES (11);
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-my $result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $initial_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -268,9 +397,13 @@ my $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $initial_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $initial_check);
-is( $result, '1|{1,2,3}
+my $initial_result = '1|{1,2,3}
2|{2,3,1}
3|{3,2,1}
4|{4,3,2}
@@ -321,6 +454,7 @@ e|{d,NULL}
(4,"{c,b,d}",4)|{"(4,\"{c,b,d}\",4)"}
(5,"{c,NULL,b}",)|{"(5,\"{c,e,b}\",1)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[1,11)
2|[2,21)
3|[3,31)
@@ -334,8 +468,10 @@ e|{d,NULL}
1|"a"=>"1"
2|"zzz"=>"foo"
3|"123"=>"321"
-4|"yellow horse"=>"moaned"',
- 'check replicated inserts on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $initial_result, 'check replicated inserts on subscriber');
+is( $result_binary, $initial_result, 'check replicated inserts on subscriber in binary');
# Run batch of updates
$node_publisher->safe_psql(
@@ -370,10 +506,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $update_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -389,9 +524,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
+
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $update_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $update_check);
-is( $result, '1|{4,5,6}
+my $update_result = '1|{4,5,6}
2|{2,3,1}
3|{3,2,1}
4|{4,5,6,1}
@@ -442,6 +581,7 @@ e|{e,d}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")",NULL}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
1|[100,1001)
2|[2,21)
3|[3,31)
@@ -455,8 +595,10 @@ e|{e,d}
1|"updated"=>"value"
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated updates on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $update_result, 'check replicated updates on subscriber');
+is( $result_binary, $update_result, 'check replicated updates on subscriber in binary');
# Run batch of deletes
$node_publisher->safe_psql(
@@ -490,10 +632,9 @@ $node_publisher->safe_psql(
));
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
-# Check the data on subscriber
-$result = $node_subscriber->safe_psql(
- 'postgres', qq(
+my $delete_check = qq(
SET timezone = '+2';
SELECT a, b FROM tst_one_array ORDER BY a;
SELECT a, b, c, d FROM tst_arrays ORDER BY a;
@@ -509,9 +650,13 @@ $result = $node_subscriber->safe_psql(
SELECT a, b FROM tst_range ORDER BY a;
SELECT a, b, c FROM tst_range_array ORDER BY a;
SELECT a, b FROM tst_hstore ORDER BY a;
-));
+);
-is( $result, '3|{3,2,1}
+# Check the data on subscribers
+$result = $node_subscriber->safe_psql('postgres', $delete_check);
+$result_binary = $node_subscriber_binary->safe_psql('postgres', $delete_check);
+
+my $delete_result = '3|{3,2,1}
4|{4,5,6,1}
5|{4,5,6,1}
{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
@@ -540,26 +685,58 @@ e|{e,d}
(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+("(2,b,2)","{""(2,b,2)"",""(3,c,3)""}",b,"{b,c,NULL,d}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
2|"updated"=>"value"
3|"also"=>"updated"
-4|"yellow horse"=>"moaned"',
- 'check replicated deletes on subscriber');
+4|"yellow horse"=>"moaned"';
+
+is( $result, $delete_result, 'check replicated deletes on subscriber');
+is( $result_binary, $delete_result, 'check replicated deletes on subscriber in binary');
# Test a domain with a constraint backed by a SQL-language function,
# which needs an active snapshot in order to operate.
$node_publisher->safe_psql('postgres',
- "INSERT INTO tst_dom_constr VALUES (11)");
+ "INSERT INTO tst_dom_constr VALUES (12)");
$node_publisher->wait_for_catchup('tap_sub');
+$node_publisher->wait_for_catchup('tap_sub_binary');
$result =
$node_subscriber->safe_psql('postgres',
"SELECT sum(a) FROM tst_dom_constr");
-is($result, '21', 'sql-function constraint on domain');
+is($result, '33', 'sql-function constraint on domain');
+
+$result_binary =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT sum(a) FROM tst_dom_constr");
+is($result_binary, '33', 'sql-function constraint on domain');
+
+# Test whether subscriber fails in case of column type mismatch
+$ddl = qq(
+ CREATE TABLE tst_binary_mismatch (a int);
+ INSERT INTO tst_binary_mismatch VALUES (1);
+);
+$node_publisher->safe_psql('postgres', $ddl);
+
+$ddl = qq(
+ CREATE TABLE tst_binary_mismatch (a bigint);
+);
+$node_subscriber->safe_psql('postgres', $ddl);
+$node_subscriber_binary->safe_psql('postgres', $ddl);
+
+$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub REFRESH PUBLICATION');
+$node_subscriber_binary->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub_binary REFRESH PUBLICATION');
+
+# Binary enabled subscription should fail
+$node_subscriber_binary->wait_for_log("ERROR: insufficient data left in message");
+
+# Binary disabled subscription should succeed
+$node_publisher->wait_for_catchup('tap_sub');
$node_subscriber->stop('fast');
+$node_subscriber_binary->stop('fast');
$node_publisher->stop('fast');
done_testing();
diff --git a/src/test/subscription/t/014_binary.pl b/src/test/subscription/t/014_binary.pl
index 8d8b35721f..395efee1c7 100644
--- a/src/test/subscription/t/014_binary.pl
+++ b/src/test/subscription/t/014_binary.pl
@@ -36,6 +36,16 @@ my $ddl = qq(
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
+# Insert some content and make sure it's synced to subscriber
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ INSERT INTO public.test_arrays (a, b, c) VALUES
+ ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}');
+
+ INSERT INTO public.test_numerical (a, b, c, d) VALUES
+ (1, 1.2, 1.3, 10);
+ ));
+
# Configure logical replication
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tpub FOR ALL TABLES");
@@ -48,27 +58,35 @@ $node_subscriber->safe_psql('postgres',
# Ensure nodes are in sync with each other
$node_subscriber->wait_for_subscription_sync($node_publisher, 'tsub');
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
+
+is( $result, '1|1.2|1.3|10
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}', 'check syned data on subscriber');
+
# Insert some content and make sure it's replicated across
$node_publisher->safe_psql(
'postgres', qq(
INSERT INTO public.test_arrays (a, b, c) VALUES
- ('{1,2,3}', '{1.1, 1.2, 1.3}', '{"one", "two", "three"}'),
('{3,1,2}', '{1.3, 1.1, 1.2}', '{"three", "one", "two"}');
INSERT INTO public.test_numerical (a, b, c, d) VALUES
- (1, 1.2, 1.3, 10),
(2, 2.2, 2.3, 20),
(3, 3.2, 3.3, 30);
));
$node_publisher->wait_for_catchup('tsub');
-my $result = $node_subscriber->safe_psql('postgres',
- "SELECT a, b, c, d FROM test_numerical ORDER BY a");
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT a, b, c, d FROM test_numerical ORDER BY a;
+ SELECT a, b, c FROM test_arrays ORDER BY a;");
is( $result, '1|1.2|1.3|10
2|2.2|2.3|20
-3|3.2|3.3|30', 'check replicated data on subscriber');
+3|3.2|3.3|30
+{1,2,3}|{1.1,1.2,1.3}|{one,two,three}
+{3,1,2}|{1.3,1.1,1.2}|{three,one,two}', 'check replicated data on subscriber');
# Test updates as well
$node_publisher->safe_psql(
--
2.25.1
^ permalink raw reply [nested|flat] 18+ messages in thread
end of thread, other threads:[~2022-11-14 12:07 UTC | newest]
Thread overview: 18+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2019-07-28 13:59 [PATCH 4/4] Consider incremental sort paths in additional places Tomas Vondra <[email protected]>
2019-07-28 13:59 [PATCH 5/8] Consider incremental sort paths in additional places Tomas Vondra <[email protected]>
2019-07-28 13:59 [PATCH 4/5] Consider incremental sort paths in additional places Tomas Vondra <[email protected]>
2022-08-10 15:03 Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-08-11 02:03 ` Re: Allow logical replication to copy tables in binary format Euler Taveira <[email protected]>
2022-08-11 11:04 ` Re: Allow logical replication to copy tables in binary format Amit Kapila <[email protected]>
2022-08-11 13:26 ` Re: Allow logical replication to copy tables in binary format Euler Taveira <[email protected]>
2022-08-11 13:46 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-08-11 17:15 ` Re: Allow logical replication to copy tables in binary format Euler Taveira <[email protected]>
2022-08-15 17:03 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-09-16 13:51 ` RE: Allow logical replication to copy tables in binary format [email protected] <[email protected]>
2022-10-03 11:50 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-10-12 01:36 ` RE: Allow logical replication to copy tables in binary format [email protected] <[email protected]>
2022-11-14 12:07 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-09-22 03:22 ` RE: Allow logical replication to copy tables in binary format [email protected] <[email protected]>
2022-09-01 22:25 ` Re: Allow logical replication to copy tables in binary format Andres Freund <[email protected]>
2022-09-07 11:51 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
2022-09-14 16:50 ` Re: Allow logical replication to copy tables in binary format Melih Mutlu <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox