Slow performance with multiple joins

[SStorm]

CrateDB version

4.6.4

CrateDB setup information

Number of nodes: 3
Heap: 4gb
CPU Cores: 2

Observed behavior

SELECT r.name as region, count(c.id) 
FROM core.regions r 
LEFT JOIN core.projects p ON (r.name = p.region) 
LEFT JOIN core.clusters c ON (c.project_id = p.id) 
WHERE (c.is_deleted is null or c.is_deleted = false) 
AND r.name NOT LIKE '%.edge' GROUP by r.name limit 100;

Takes anywhere from 5 to 20 seconds.

All 3 tables have < 500 entries.

EXPLAIN

Eval[name AS region, count(id)]
  └ Limit[100::bigint;0]
    └ GroupHashAggregate[name | count(id)]
      └ Filter[((is_deleted IS NULL) OR (is_deleted = false))]
        └ NestedLoopJoin[LEFT | (project_id = id)]
          ├ NestedLoopJoin[RIGHT | (name = region)]
          │  ├ Rename[region, id] AS p
          │  │  └ Collect[core.projects | [region, id] | true]
          │  └ Rename[name] AS r
          │    └ Collect[core.regions | [name] | (NOT (name LIKE '%.edge'))]
          └ Rename[id, is_deleted, project_id] AS c
            └ Collect[core.clusters | [id, is_deleted, project_id] | true]

Steps to Reproduce

CREATE TABLE IF NOT EXISTS "core"."regions" (
   "description" TEXT,
   "name" TEXT,
   "agent_url" TEXT,
   "prometheus_url" TEXT,
   "basic_auth_password" TEXT,
   "organization_id" TEXT,
   "aws_region" TEXT,
   "aws_bucket" TEXT,
   "deprecated" BOOLEAN,
   "provider" TEXT,
   "status" TEXT,
   "last_seen" TIMESTAMP WITH TIME ZONE,
   "dc" OBJECT(STRICT) AS (
      "created" TIMESTAMP WITH TIME ZONE,
      "modified" TIMESTAMP WITH TIME ZONE
   ),
   "cert_type" TEXT,
   PRIMARY KEY ("name")
)
CLUSTERED BY ("name") INTO 1 SHARDS
WITH (
   "allocation.max_retries" = 5,
   "blocks.metadata" = false,
   "blocks.read" = false,
   "blocks.read_only" = false,
   "blocks.read_only_allow_delete" = false,
   "blocks.write" = false,
   codec = 'default',
   column_policy = 'strict',
   "mapping.total_fields.limit" = 1000,
   max_ngram_diff = 1,
   max_shingle_diff = 3,
   number_of_replicas = '1',
   refresh_interval = 1000,
   "routing.allocation.enable" = 'all',
   "routing.allocation.total_shards_per_node" = -1,
   "store.type" = 'fs',
   "translog.durability" = 'REQUEST',
   "translog.flush_threshold_size" = 536870912,
   "translog.sync_interval" = 5000,
   "unassigned.node_left.delayed_timeout" = 60000,
   "warmer.enabled" = true,
   "write.wait_for_active_shards" = '1'
)

CREATE TABLE IF NOT EXISTS "core"."projects" (
   "dc" OBJECT(DYNAMIC) AS (
      "created" BIGINT,
      "modified" BIGINT
   ),
   "friendly_name" TEXT,
   "id" TEXT,
   "name" TEXT,
   "organization_id" TEXT,
   "region" TEXT,
   "is_deleted" BOOLEAN,
   PRIMARY KEY ("id")
)
CLUSTERED BY ("id") INTO 1 SHARDS
WITH (
   "allocation.max_retries" = 5,
   "blocks.metadata" = false,
   "blocks.read" = false,
   "blocks.read_only" = false,
   "blocks.read_only_allow_delete" = false,
   "blocks.write" = false,
   codec = 'default',
   column_policy = 'strict',
   "mapping.total_fields.limit" = 1000,
   max_ngram_diff = 1,
   max_shingle_diff = 3,
   number_of_replicas = '1',
   refresh_interval = 1000,
   "routing.allocation.enable" = 'all',
   "routing.allocation.total_shards_per_node" = -1,
   "store.type" = 'fs',
   "translog.durability" = 'REQUEST',
   "translog.flush_threshold_size" = 536870912,
   "translog.sync_interval" = 5000,
   "unassigned.node_left.delayed_timeout" = 60000,
   "warmer.enabled" = true,
   "write.wait_for_active_shards" = '1'
)

CREATE TABLE IF NOT EXISTS "core"."clusters" (
   "id" TEXT,
   "name" TEXT,
   "username" TEXT,
   "dc" OBJECT(STRICT) AS (
      "created" TIMESTAMP WITH TIME ZONE,
      "modified" TIMESTAMP WITH TIME ZONE
   ),
   "num_nodes" INTEGER,
   "hardware_specs" OBJECT(STRICT) AS (
      "cpus_per_node" REAL,
      "disk_size_per_node" TEXT,
      "disk_size_per_node_bytes" BIGINT,
      "disk_type" TEXT,
      "disks_per_node" INTEGER,
      "heap_ratio" REAL,
      "heap_size" TEXT,
      "heap_size_bytes" BIGINT,
      "memory_per_node" TEXT,
      "memory_per_node_bytes" BIGINT
   ),
   "crate_version" TEXT,
   "project_id" TEXT,
   "product_name" TEXT,
   "product_tier" TEXT,
   "product_unit" INTEGER,
   "channel" TEXT,
   "origin" TEXT,
   "is_deleted" BOOLEAN DEFAULT false,
   "cluster_version" SMALLINT,
   "subscription_id" TEXT,
   "external_ip" TEXT,
   "deletion_protected" BOOLEAN,
   "ip_whitelist" ARRAY(OBJECT(IGNORED)),
   "status" TEXT,
   "last_seen" TIMESTAMP WITH TIME ZONE,
   PRIMARY KEY ("id")
)
CLUSTERED BY ("id") INTO 1 SHARDS
WITH (
   "allocation.max_retries" = 5,
   "blocks.metadata" = false,
   "blocks.read" = false,
   "blocks.read_only" = false,
   "blocks.read_only_allow_delete" = false,
   "blocks.write" = false,
   codec = 'default',
   column_policy = 'strict',
   "mapping.total_fields.limit" = 1000,
   max_ngram_diff = 1,
   max_shingle_diff = 3,
   number_of_replicas = '1',
   refresh_interval = 1000,
   "routing.allocation.enable" = 'all',
   "routing.allocation.total_shards_per_node" = -1,
   "store.type" = 'fs',
   "translog.durability" = 'REQUEST',
   "translog.flush_threshold_size" = 536870912,
   "translog.sync_interval" = 5000,
   "unassigned.node_left.delayed_timeout" = 60000,
   "write.wait_for_active_shards" = '1'
)