onderkalaci

Distributed Query Planning with Examples in Citus (as of Citus 12.1)

This part of the documentation aims to provide a comprehensive understanding of how Citus handles distributed query planning. We will use a set of realistic tables to demonstrate various query queries. Through these examples, we hope to offer a step-by-step guide on how Citus chooses to plan queries.

Non-Goal: This document does not aim to provide the high level design of the distributed planner, but instead the lowest levels of details possible. For high-level flow, go to https://postgresconf.org/system/events/document/000/000/233/Distributing_Queries_the_Citus_Way.pdf

onderkalaci

 WITH joins_only AS (SELECT left_tablename, p1.attname, right_tablename, p2.attname, call_count  FROM public.equi_filters, pg_attribute p1, pg_attribute p2 WHERE right_tablename != 'NULL' AND equi_filters.left_tablename::text::regclass = p1.attrelid AND equi_filters.right_tablename::text::regclass = p2.attrelid AND p1.attnum = left_attrno AND p2.attnum = right_attrname) SELECT * FROM joins_only;



WITH only_filters AS (SELECT left_tablename, attname,call_count FROM pg_attribute, equi_filters WHERE right_tablename = 'NULL' AND  equi_filters.left_tablename::text::regclass = attrelid AND attnum=left_attrno) SELECT * FROM only_filters;

onderkalaci

onderkalaci

CREATE TABLE test(a int, b int, c int);
ALTER TABLE test REPLICA IDENTITY FULL;
INSERT INTO test SELECT i,i,i FROM generate_series(0,1000000)i;

CREATE PUBLICATION pub_test_1 FOR TABLE test(a,b) WHERE (a%2=0);


-- on the target, same table with an index
CREATE TABLE test(a int, b int, c int);
CREATE INDEX i1 ON test(a);

onderkalaci

#!/usr/bin/python

import threading
import time
import os
import random

exitFlag = 0

class myThread (threading.Thread):

onderkalaci

-- Create a distributed table on the coordinator
CREATE TABLE test_distributed(key bigint PRIMARY KEY, data jsonb);
SELECT create_distributed_table('test_distributed', 'key');

-- connect to any worker and show table is there
psql -h localhost -p 9700 postgres
\d+ test_distributed

-- Insert from one worker,
INSERT INTO test_distributed VALUES (1, '{"value": "100"}');

onderkalaci

-- very basic test
-- single column table on the source with rep. identity full
CREATE TABLE test(a int);
ALTER TABLE test REPLICA IDENTITY FULL;
INSERT INTO test SELECT i FROM generate_series(0,1000000)i;

CREATE PUBLICATION pub_test_1 FOR ALL TABLES;


-- on the target, same table with an index

onderkalaci

CREATE OR REPLACE FUNCTION generate_text(len int)
RETURNS text AS $function$
SELECT 'f70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb390aca6ccbd81115087d9def70f1c1eb39

onderkalaci

-- weird error
CREATE TEXT SEARCH CONFIGURATION en ( parser = default );
ALTER TEXT SEARCH CONFIGURATION en ALTER MAPPING FOR host, host WITH simple;
ERROR:  duplicate key value violates unique constraint "pg_ts_config_map_index"
DETAIL:  Key (mapcfg, maptokentype, mapseqno)=(25798, 6, 1) already exists.


-- same error, sometimes we get "ERROR:  tuple already updated by self"
CREATE TEXT SEARCH CONFIGURATION dutch ( copy = dutch );

onderkalaci

-- prevent sending types /schema
SET citus.enable_ddl_propagation TO OFF;

CREATE SCHEMA local_schema;
SET search_path TO local_schema;
CREATE TYPE dup_result AS (f1 int, f2 text);
CREATE FUNCTION dup(int) RETURNS dup_result
    AS $$ SELECT $1, CAST($1 AS text) || ' is text' $$