Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,10 @@ describe("buildLargestClusterReplicaQuery", () => {
replica_id TEXT NOT NULL,
time_ns bigint
);
> CREATE TABLE mz_cluster_replica_sizes (
size TEXT NOT NULL,
processes uint8 NOT NULL
);
`);

const {
Expand Down Expand Up @@ -85,6 +89,31 @@ describe("buildLargestClusterReplicaQuery", () => {
expect(result.size).toBe("scale=1,workers=1,mem=8GiB");
expect(result.heapLimit).toBe("8589934592");

const {
rows: [named],
} = await executeSqlHttp(
buildLargestClusterReplicaQuery(cluster.id, "r1").compile(),
{
sessionVariables: {
search_path: mockedSearchPath,
cluster: QUICKSTART_CLUSTER,
},
},
);
expect(named.name).toBe("r1");
expect(named.heapLimit).toBe("4294967296");

// Object sizes cover every process of a replica, so its heap limit does
// too. Four 4GiB processes outrank one 8GiB process.
await client.query(`
INSERT INTO internal_test.mz_cluster_replica_sizes VALUES
('scale=1,workers=1,mem=4GiB', 4)
`);
const multiProcess = await fetchLargest();
expect([r1.name, r3.name]).toContain(multiProcess.name);
expect(multiProcess.heapLimit).toBe("17179869184");
await client.query(`DELETE FROM internal_test.mz_cluster_replica_sizes`);

// A hydrated replica is preferred over a larger, still-hydrating one
// (its object is present but has no hydration time yet).
await client.query(`
Expand Down
86 changes: 52 additions & 34 deletions console/src/api/materialize/cluster/largestClusterReplica.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,44 +13,61 @@ import { sql } from "kysely";
import { executeSqlV2, queryBuilder } from "~/api/materialize";
import { buildClusterReplicaHeapMetricsTable } from "~/api/materialize/expressionBuilders";

export function buildLargestClusterReplicaQuery(clusterId: string) {
return (
queryBuilder
.selectFrom("mz_cluster_replicas as cr")
.leftJoin(
buildClusterReplicaHeapMetricsTable().as("crhm"),
"crhm.replica_id",
"cr.id",
)
// Prefer a fully-hydrated replica, whose reported sizes are stable.
// Joining mz_compute_hydration_times before aggregating lets its
// replica_id index bound the scan to this cluster's replicas, unlike an
// aggregation over the whole environment's mz_hydration_statuses.
.leftJoin("mz_compute_hydration_times as ht", (join) =>
join
.onRef("ht.replica_id", "=", "cr.id")
// System introspection indexes may never report hydrated.
.on("ht.object_id", "not like", "s%"),
)
.select([
"cr.name",
"cr.size",
sql<string | null>`crhm.heap_limit::text`.as("heapLimit"),
])
.where("cr.cluster_id", "=", clusterId)
.groupBy(["cr.id", "cr.name", "cr.size", "crhm.heap_limit"])
// Prefer a replica whose compute objects are all hydrated, so its
// reported sizes are stable. time_ns is NULL until an object hydrates. A
// replica with no hydration rows, such as one just added, has bool_and
// over an all-NULL row and so sorts last. Then order by heap limit.
.orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`)
.orderBy(sql`crhm.heap_limit DESC NULLS LAST`)
.limit(1)
);
// `heap_limit` is per process, but the object sizes divided by it cover every
// process of the replica.
const replicaHeapLimit = sql`(crhm.heap_limit * COALESCE(crs.processes, 1))`;

export function buildLargestClusterReplicaQuery(
clusterId: string,
replicaName?: string,
) {
const query = queryBuilder
.selectFrom("mz_cluster_replicas as cr")
.leftJoin(
buildClusterReplicaHeapMetricsTable().as("crhm"),
"crhm.replica_id",
"cr.id",
)
// Prefer a fully-hydrated replica, whose reported sizes are stable.
// Joining mz_compute_hydration_times before aggregating lets its
// replica_id index bound the scan to this cluster's replicas, unlike an
// aggregation over the whole environment's mz_hydration_statuses.
.leftJoin("mz_compute_hydration_times as ht", (join) =>
join
.onRef("ht.replica_id", "=", "cr.id")
// System introspection indexes may never report hydrated.
.on("ht.object_id", "not like", "s%"),
)
.leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size")
.select([
"cr.name",
"cr.size",
sql<string | null>`${replicaHeapLimit}::text`.as("heapLimit"),
])
.where("cr.cluster_id", "=", clusterId)
.groupBy([
"cr.id",
"cr.name",
"cr.size",
"crhm.heap_limit",
"crs.processes",
])
// Prefer a replica whose compute objects are all hydrated, so its
// reported sizes are stable. time_ns is NULL until an object hydrates. A
// replica with no hydration rows, such as one just added, has bool_and
// over an all-NULL row and so sorts last. Then order by heap limit.
.orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`)
.orderBy(sql`${replicaHeapLimit} DESC NULLS LAST`)
.limit(1);
return replicaName === undefined
? query
: query.where("cr.name", "=", replicaName);
}

export type LargestClusterReplicaParams = {
clusterId: string;
/** Reads this replica, with the same heap limit, instead of the largest. */
replicaName?: string;
};

/**
Expand All @@ -67,6 +84,7 @@ export async function fetchLargestClusterReplica({
}) {
const compiledQuery = buildLargestClusterReplicaQuery(
params.clusterId,
params.replicaName,
).compile();
return executeSqlV2({
queries: compiledQuery,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import {
} from "~/api/materialize";

export type LargestMaintainedQueriesParams = {
replicaHeapLimit: number;
replicaHeapLimit: number | null;
limit: number;
clusterName: string;
replicaName: string;
Expand Down Expand Up @@ -87,7 +87,9 @@ export function buildLargestMaintainedQueriesQuery({
})
// Filter out transient dataflows
.where("ce.export_id", "not like", "t%")
.orderBy("memoryPercentage", sql`desc NULLS LAST`)
// Order by size: without a heap limit, memoryPercentage is NULL on every
// row.
.orderBy("size", sql`desc NULLS LAST`)
.limit(() => sql.raw(limit.toString()))
);
}
Expand Down Expand Up @@ -119,7 +121,7 @@ export async function fetchLargestMaintainedQueries({
export type LargestMaintainedObjectSizesParams = {
clusterId: string;
replicaName: string;
replicaHeapLimit: number;
replicaHeapLimit: number | null;
limit: number;
};

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,7 @@ vi.mock("~/hooks/useFlags", () => ({
}));
// The page's cards have their own tests and queries.
vi.mock("./ResourceUsage/ResourceUsage", () => ({ ResourceUsage: () => null }));
vi.mock("~/platform/clusters/LargestMaintainedQueries", () => ({
default: () => null,
}));
vi.mock("./MemoryByObject", () => ({ MemoryByObject: () => null }));

const renderClusterMetrics = () =>
renderComponent(
Expand Down
15 changes: 7 additions & 8 deletions console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

import { Box, HStack, Text, VStack } from "@chakra-ui/react";
import { HStack, Text, VStack } from "@chakra-ui/react";
import { sort } from "d3";
import React from "react";
import { useParams } from "react-router-dom";
Expand All @@ -22,10 +22,10 @@ import { MainContentContainer } from "~/layouts/BaseLayout";
import { ClusterInfoBox } from "~/platform/clusters/ClusterOverview";
import { ClusterParams } from "~/platform/clusters/ClusterRoutes";
import { CLUSTERS_FETCH_ERROR_MESSAGE } from "~/platform/clusters/constants";
import LargestMaintainedQueries from "~/platform/clusters/LargestMaintainedQueries";
import { useAllClusters } from "~/store/allClusters";
import { assert } from "~/util";

import { MemoryByObject } from "./MemoryByObject";
import { ResourceUsage } from "./ResourceUsage/ResourceUsage";
import { useSelectedReplica } from "./useSelectedReplica";

Expand Down Expand Up @@ -78,12 +78,11 @@ const ClusterMetricsContent = ({ cluster }: { cluster: Cluster }) => {
timePeriodMinutes={timePeriodMinutes}
selectedReplicaId={selectedReplica?.id}
/>
<Box width="100%">
<LargestMaintainedQueries
clusterId={cluster.id}
clusterName={cluster.name}
/>
</Box>
<MemoryByObject
clusterId={cluster.id}
clusterName={cluster.name}
replicaName={selectedReplica?.name}
/>
</VStack>
</MainContentContainer>
);
Expand Down
Loading
Loading