Citus Documentation

Welcome to the documentation for Citus 5.1! Citus horizontally scales PostgreSQL across commodity servers using sharding and replication. Its query engine parallelizes incoming SQL queries across these servers to enable real-time responses on large datasets.

The documentation explains how you can install Citus and then provides instructions to design, build, query, and maintain your Citus cluster. It also includes a Reference section which provides quick information on several topics.

About Citus

• What is Citus?
  • When to Use Citus
  • Considerations for Use
  • When Citus is Inappropriate
• Architecture
  • Master / Worker Nodes
  • Logical Sharding
  • Metadata Tables
  • Query Processing
  • Failure Handling

Tutorials

• Start Demo Cluster
• Hash-Distributed Data
• Append-Distributed Data

Installation

• Requirements
• Single-Machine Cluster
  • OS X
  • Fedora, CentOS, or Red Hat
  • Ubuntu or Debian
  • Docker
• Multi-Machine Cluster
  • Amazon Web Services
    • Managed Citus Cloud Deployment
    • Manual CloudFormation Deployment
  • Multi-node setup on Fedora, CentOS, or Red Hat
    • Steps to be executed on all nodes
    • Steps to be executed on the master node
  • Multi-node setup on Ubuntu or Debian
    • Steps to be executed on all nodes
    • Steps to be executed on the master node

Distributed Tables

• Working with distributed tables
• Hash Distribution
• Append Distribution
• Querying Distributed Tables
• PostgreSQL extensions

Performance

• Citus Query Processing
  • Distributed Query Planner
  • Distributed Query Executor
    • Real-time Executor
    • Task Tracker Executor
  • PostgreSQL planner and executor
• Scaling Out Data Ingestion
  • Hash Distributed Tables
    • Real-time Inserts (0-50k/s)
    • Real-time Updates (0-50k/s)
    • Bulk Copy (100-200k/s)
    • Masterless Citus (50k/s-500k/s)
  • Append Distributed Tables
    • Master Node Bulk Ingestion (100k/s-200k/s)
    • Worker Node Bulk Ingestion (100k/s-1M/s)
  • Pre-processing Data in Citus
• Query Performance Tuning
  • Table Distribution and Shards
  • PostgreSQL tuning
  • Scaling Out Performance
  • Distributed Query Performance Tuning
    • General
    • Advanced
      • Task Assignment Policy
      • Intermediate Data Transfer Format
      • Real Time Executor
      • Task Tracker Executor

Cloud

• Citus Overview
• Features
• Support and Billing

Technical Solutions

• Real Time Dashboards
  • Running It Yourself
  • Data Model
  • Rollups
  • Expiring Old Data
  • Approximate Distinct Counts
  • Unstructured Data with JSONB
  • Resources

Administration

• Cluster Management
• Upgrading Citus
• Transitioning From PostgreSQL to Citus

Reference

• Citus SQL Language Reference
• User Defined Functions Reference
  • Table and Shard DDL
    • master_create_distributed_table
      • Arguments
      • Return Value
      • Example
    • master_create_worker_shards
      • Arguments
      • Return Value
      • Example
    • master_create_empty_shard
      • Arguments
      • Return Value
      • Example
    • master_append_table_to_shard
      • Arguments
      • Return Value
      • Example
    • master_apply_delete_command
      • Arguments
      • Return Value
      • Example
  • Metadata / Configuration Information
    • master_get_active_worker_nodes
      • Arguments
      • Return Value
      • Example
    • master_get_table_metadata
      • Arguments
      • Return Value
      • Example
  • Cluster Management And Repair Functions
    • master_copy_shard_placement
      • Arguments
      • Return Value
      • Example
    • rebalance_table_shards
      • Arguments
      • Return Value
      • Example
    • replicate_table_shards
      • Arguments
      • Return Value
      • Examples
• Metadata Tables Reference
  • Partition table
  • Shard table
    • Shard Storage Types
  • Shard placement table
    • Shard Placement States
• Configuration Reference
  • Node configuration
    • pg_worker_list.conf
    • citus.max_worker_nodes_tracked (integer)
  • Data Loading
    • citus.multi_shard_commit_protocol (enum)
    • citus.shard_replication_factor (integer)
    • citus.shard_max_size (integer)
  • Planner Configuration
    • citus.large_table_shard_count (integer)
    • citus.limit_clause_row_fetch_count (integer)
    • citus.count_distinct_error_rate (floating point)
    • citus.task_assignment_policy (enum)
  • Intermediate Data Transfer Format
    • citus.binary_worker_copy_format (boolean)
    • citus.binary_master_copy_format (boolean)
  • Executor Configuration
    • citus.all_modifications_commutative
    • citus.remote_task_check_interval (integer)
    • citus.task_executor_type (enum)
    • Real-time executor configuration
      • max_connections (integer)
      • max_files_per_process (integer)
    • Task tracker executor configuration
      • citus.task_tracker_delay (integer)
      • citus.max_tracked_tasks_per_node (integer)
      • citus.max_assign_task_batch_size (integer)
      • citus.max_running_tasks_per_node (integer)
      • citus.partition_buffer_size (integer)
    • Explain output
      • citus.explain_all_tasks (boolean)

FAQ

• Frequently Asked Questions
  • Can I create primary keys on distributed tables?
  • How do I add nodes to an existing Citus cluster?
  • How do I change the shard count for a hash partitioned table?
  • How does citus hadle failure of a worker node?
  • How does Citus handle failover of the master node?
  • How do I ingest the results of a query into a distributed table?
  • Can I join distributed and non-distributed tables together in the same query?
  • Are there any PostgreSQL features not supported by CitusDB?
  • How do I choose the shard count when I hash-partition my data?
  • How does citus support count(distinct) queries?