diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineConnectionReview.tsx b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineConnectionReview.tsx new file mode 100644 index 00000000000..dac4bb4df77 --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineConnectionReview.tsx @@ -0,0 +1,69 @@ +import { useWatch, type Control } from 'react-hook-form' +import { Input } from 'ui' +import { FormItemLayout } from 'ui-patterns/form/FormItemLayout/FormItemLayout' + +import { + DEFAULT_CONNECTION_POOL_SIZE, + DEFAULT_DUCKLAKE_POOL_SIZE, +} from '../DestinationPanel/DestinationForm/DestinationForm.constants' +import type { DestinationPanelSchemaType } from '../DestinationPanel/DestinationForm/DestinationForm.schema' +import type { DestinationType } from '../DestinationPanel/DestinationPanel.types' + +export const PipelineConnectionReview = ({ + type, + control, +}: { + type: DestinationType + control: Control +}) => { + const [pool, staleness, lakePool, role] = useWatch({ + control, + name: ['connectionPoolSize', 'maxStalenessMins', 'ducklakePoolSize', 'snowflakeRole'], + }) + const rows: { label: string; value: string; description: string }[] = [] + if (type === 'BigQuery') { + if (typeof pool === 'number' && pool !== DEFAULT_CONNECTION_POOL_SIZE) + rows.push({ + label: 'Connection pool size', + value: String(pool), + description: 'Number of BigQuery connections used for destination writes.', + }) + if (typeof staleness === 'number') + rows.push({ + label: 'Maximum staleness', + value: `${staleness} minutes`, + description: 'Maximum age of BigQuery query results for newly created or recreated tables.', + }) + } + if ( + type === 'DuckLake' && + typeof lakePool === 'number' && + lakePool !== DEFAULT_DUCKLAKE_POOL_SIZE + ) + rows.push({ + label: 'Pool size', + value: String(lakePool), + description: 'Maximum concurrent connections this pipeline opens to the catalog.', + }) + if (type === 'Snowflake' && role?.trim()) + rows.push({ + label: 'Role', + value: role, + description: 'Snowflake role used for SQL requests.', + }) + return ( + <> + {rows.map(({ label, value, description }) => ( + + + + ))} + + ) +} diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.test.tsx b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.test.tsx new file mode 100644 index 00000000000..41a92ff6eea --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.test.tsx @@ -0,0 +1,198 @@ +import { fireEvent, screen } from '@testing-library/react' +import { useForm } from 'react-hook-form' +import { Button } from 'ui' +import { describe, expect, it } from 'vitest' + +import type { DestinationPanelSchemaType } from '../DestinationPanel/DestinationForm/DestinationForm.schema' +import type { DestinationType } from '../DestinationPanel/DestinationPanel.types' +import { PipelineConnectionReview } from './PipelineConnectionReview' +import { PipelineDataReview } from './PipelineDataReview' +import { describeBigQueryTableLayout } from './PipelineReviewSummary.utils' +import type { ReplicationPublicationData } from '@/data/replication/publication-query' +import { customRender } from '@/tests/lib/custom-render' + +const publication: ReplicationPublicationData = { + name: 'analytics', + config: { type: 'all_tables', operations: ['insert'], publish_via_partition_root: false }, + tables: [{ id: 101, schema: 'public', name: 'orders', kind: 'table', partition_parent_id: null }], +} +const TestReview = ({ + type = 'BigQuery', + values = {}, + available = true, +}: { + type?: DestinationType + values?: Partial + available?: boolean +}) => { + const form = useForm({ + defaultValues: { + publicationName: 'analytics', + tableSyncCopyMode: 'include_all_tables', + tableSyncCopyTableIds: [], + maxFillMs: 10000, + maxTableSyncWorkers: 4, + maxCopyConnectionsPerTable: 4, + connectionPoolSize: 4, + ducklakePoolSize: 4, + invalidatedSlotBehavior: 'error', + ...values, + }, + }) + return ( + <> + + + + + + ) +} + +describe('PipelineDataReview', () => { + it('shows data defaults without unchanged advanced settings and reacts to changes', () => { + customRender() + expect(screen.getByLabelText('Publication')).toHaveValue('analytics') + expect(screen.getByLabelText('Publication tables')).toHaveValue('1 table') + expect(screen.getByLabelText('Initial sync')).toHaveValue('All tables') + expect(screen.queryByLabelText('Table sync workers')).not.toBeInTheDocument() + expect(screen.queryByText('Custom connection settings')).not.toBeInTheDocument() + fireEvent.click(screen.getByText('Change workers')) + expect(screen.getByLabelText('Table sync workers')).toHaveValue('8') + expect( + screen.getByText('Maximum number of tables synced at the same time.') + ).toBeInTheDocument() + fireEvent.click(screen.getByText('Change publication')) + expect(screen.getByLabelText('Publication tables')).toHaveValue('Unavailable') + }) + it.each(['include_tables', 'skip_tables'] as const)( + 'lists selected publication tables for %s', + (mode) => { + customRender( + + ) + expect( + screen.getByLabelText(mode === 'include_tables' ? 'Tables to include' : 'Tables to exclude') + ).toHaveValue('public.orders') + } + ) + it('shows custom replication and BigQuery settings and only configured layouts', () => { + customRender( + + ) + expect(screen.getByLabelText('Batch wait time')).toHaveValue('500 milliseconds') + expect( + screen.getByText('Maximum time before sending a partially filled batch.') + ).toBeInTheDocument() + expect( + screen.getByText( + 'Maximum number of source connections used to sync existing rows for each table.' + ) + ).toBeInTheDocument() + expect( + screen.getByText( + 'If the replication slot becomes invalid, the pipeline replaces destination tables and runs a new, billable initial sync.' + ) + ).toBeInTheDocument() + expect(screen.getByLabelText('Connection pool size')).toHaveValue('6') + expect(screen.getByLabelText('Maximum staleness')).toHaveValue('0 minutes') + expect(screen.getByText('public.orders')).toBeInTheDocument() + expect(screen.getByText('No partitioning; Cluster by id')).toBeInTheDocument() + expect(screen.queryByText('Table 999')).not.toBeInTheDocument() + }) + it.each([ + { type: 'DuckLake', values: { ducklakePoolSize: 6 }, label: 'Pool size', value: '6' }, + { type: 'Snowflake', values: { snowflakeRole: 'ANALYST' }, label: 'Role', value: 'ANALYST' }, + ] satisfies { + type: DestinationType + values: Partial + label: string + value: string + }[])('shows custom settings for $type', ({ type, values, label, value }) => { + customRender() + expect(screen.getByLabelText(label)).toHaveValue(value) + expect(screen.queryByLabelText('Connection pool size')).not.toBeInTheDocument() + }) + it('marks unavailable publication data without inventing a table count', () => { + customRender() + expect(screen.getByLabelText('Publication tables')).toHaveValue('Unavailable') + }) + it('puts the table name and incomplete partition summary in the value column', () => { + customRender( + + ) + expect(screen.getByText('Table layout')).toBeInTheDocument() + expect( + screen.getByText( + 'Partitioning and clustering applied when destination tables are first created or reset.' + ) + ).toBeInTheDocument() + expect(screen.getByText('public.orders')).toHaveClass('break-all') + expect(screen.getByText('Partition by Not selected (month); Cluster by table_id')).toHaveClass( + 'break-words' + ) + expect(screen.queryByText('Table layout: public.orders')).not.toBeInTheDocument() + }) + + it('does not describe unavailable selections as empty or show stale layouts', () => { + const view = customRender( + + ) + expect(screen.getByLabelText('Tables to include')).toHaveValue('Unavailable') + expect(screen.queryByText('Table 999')).not.toBeInTheDocument() + view.unmount() + customRender() + expect(screen.queryByText('Table layout')).not.toBeInTheDocument() + }) + + it('describes each partition mode', () => { + expect( + describeBigQueryTableLayout({ + tableId: 1, + partitionBy: { kind: 'ingestion_time', granularity: 'day' }, + }) + ).toBe('Partition by ingestion time (day); No clustering') + expect( + describeBigQueryTableLayout({ + tableId: 1, + partitionBy: { kind: 'time_column', column: 'created_at', granularity: 'day' }, + }) + ).toBe('Partition by created_at (day); No clustering') + expect( + describeBigQueryTableLayout({ + tableId: 1, + partitionBy: { kind: 'integer_range', column: 'id', start: 0, end: 100, interval: 10 }, + }) + ).toBe('Partition by id (range 0 to 100, interval 10); No clustering') + }) +}) diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.tsx b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.tsx new file mode 100644 index 00000000000..3a713f1d545 --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDataReview.tsx @@ -0,0 +1,157 @@ +import { useWatch, type Control } from 'react-hook-form' +import { Input } from 'ui' +import { FormItemLayout } from 'ui-patterns/form/FormItemLayout/FormItemLayout' + +import * as defaults from '../DestinationPanel/DestinationForm/DestinationForm.constants' +import type { DestinationPanelSchemaType } from '../DestinationPanel/DestinationForm/DestinationForm.schema' +import { + INITIAL_SYNC_FIELD_COPY, + INITIAL_SYNC_LABELS, + PUBLICATION_FIELD_COPY, +} from '../DestinationPanel/DestinationForm/DestinationFormFieldCopy' +import type { DestinationType } from '../DestinationPanel/DestinationPanel.types' +import { describeBigQueryTableLayout } from './PipelineReviewSummary.utils' +import type { ReplicationPublicationData } from '@/data/replication/publication-query' + +export const PipelineDataReview = ({ + type, + control, + publication, +}: { + type: DestinationType + control: Control + publication?: ReplicationPublicationData +}) => { + const [publicationName, mode, ids, workers, copies, slot, batch, options] = useWatch({ + control, + name: [ + 'publicationName', + 'tableSyncCopyMode', + 'tableSyncCopyTableIds', + 'maxTableSyncWorkers', + 'maxCopyConnectionsPerTable', + 'invalidatedSlotBehavior', + 'maxFillMs', + 'tableOptions', + ], + }) + const tables = publication?.name === publicationName ? publication.tables : undefined + const rows: { label: string; value: string; description: string }[] = [ + { ...PUBLICATION_FIELD_COPY, value: publicationName || 'None selected' }, + { + label: 'Publication tables', + description: 'Tables whose ongoing changes will be replicated to this destination.', + value: tables + ? `${tables.length} ${tables.length === 1 ? 'table' : 'tables'}` + : 'Unavailable', + }, + { + ...INITIAL_SYNC_FIELD_COPY, + description: + 'Controls which publication tables copy their existing rows. Ongoing replication includes new changes from every publication table, even when initial sync is skipped.', + value: INITIAL_SYNC_LABELS[mode] ?? 'Not set', + }, + ] + if (mode === 'include_tables' || mode === 'skip_tables') { + const selected = tables?.filter((table) => ids?.includes(String(table.id))) + rows.push({ + label: mode === 'include_tables' ? 'Tables to include' : 'Tables to exclude', + description: + mode === 'include_tables' + ? 'Only these tables copy their existing rows during initial sync. Ongoing changes from every publication table are replicated.' + : 'These tables skip copying existing rows during initial sync. Ongoing changes from every publication table are replicated.', + value: tables + ? selected?.map(({ schema, name }) => `${schema}.${name}`).join(', ') || 'None selected' + : 'Unavailable', + }) + } + const addNumber = ( + label: string, + value: number | '' | undefined, + defaultValue: number, + description: string, + unit = '' + ) => { + if (typeof value === 'number' && value !== defaultValue) + rows.push({ label, value: `${value}${unit}`, description }) + } + addNumber( + 'Batch wait time', + batch, + defaults.DEFAULT_MAX_FILL_MS, + 'Maximum time before sending a partially filled batch.', + ' milliseconds' + ) + addNumber( + 'Table sync workers', + workers, + defaults.DEFAULT_MAX_TABLE_SYNC_WORKERS, + 'Maximum number of tables synced at the same time.' + ) + addNumber( + 'Initial sync connections per table', + copies, + defaults.DEFAULT_MAX_COPY_CONNECTIONS_PER_TABLE, + 'Maximum number of source connections used to sync existing rows for each table.' + ) + if (slot === 'recreate') + rows.push({ + label: 'Invalidated slot behavior', + value: 'Recreate slot', + description: + 'If the replication slot becomes invalid, the pipeline replaces destination tables and runs a new, billable initial sync.', + }) + const layouts = + type === 'BigQuery' + ? (options ?? []).filter( + (option) => + tables?.some(({ id }) => id === option.tableId) && + (option.partitionBy || option.clusterBy?.length) + ) + : [] + return ( +
+

Data

+ {rows.map((row) => ( + + ))} + {layouts.length > 0 && ( + +
+ {layouts.map((option) => { + const table = tables?.find(({ id }) => id === option.tableId) + const name = table ? `${table.schema}.${table.name}` : `Table ${option.tableId}` + return ( +
+

{name}

+

+ {describeBigQueryTableLayout(option)} +

+
+ ) + })} +
+
+ )} +
+ ) +} + +const ReviewField = ({ + label, + value, + description, +}: { + label: string + value: string + description: string +}) => ( + + + +) diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.test.tsx b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.test.tsx index 9aa609740ae..cb3dde1ef10 100644 --- a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.test.tsx +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.test.tsx @@ -114,3 +114,21 @@ describe('PipelineDestinationReview', () => { expect(screen.queryByText('Catalog URL')).not.toBeInTheDocument() }) }) + +it('includes custom settings in Connection without a separate heading', () => { + customRender( + + ) + expect(screen.getByLabelText('Connection pool size')).toHaveValue('3') + expect(screen.getByLabelText('Maximum staleness')).toHaveValue('10 minutes') + expect( + screen.getByText('Number of BigQuery connections used for destination writes.') + ).toBeInTheDocument() + expect( + screen.getByText('Maximum age of BigQuery query results for newly created or recreated tables.') + ).toBeInTheDocument() + expect(screen.queryByText('Custom connection settings')).not.toBeInTheDocument() +}) diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.tsx b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.tsx index 77245e8ec02..f5e6d126fd7 100644 --- a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.tsx +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineDestinationReview.tsx @@ -1,4 +1,4 @@ -import { Eye, EyeOff } from 'lucide-react' +import { EyeIcon, EyeOffIcon } from 'lucide-react' import { useState } from 'react' import { useWatch, type Control } from 'react-hook-form' import { Button, Input, Separator } from 'ui' @@ -23,11 +23,9 @@ import { SNOWFLAKE_DATABASE_FIELD_COPY, SNOWFLAKE_SCHEMA_FIELD_COPY, } from '../DestinationPanel/DestinationForm/DestinationFormFieldCopy' -import { - getPipelineRegionDescription, - PipelineRegionReadonly, -} from '../DestinationPanel/DestinationForm/PipelineRegionField' +import { PipelineRegionReadonly } from '../DestinationPanel/DestinationForm/PipelineRegionField' import type { DestinationType } from '../DestinationPanel/DestinationPanel.types' +import { PipelineConnectionReview } from './PipelineConnectionReview' const MaskedUrl = ({ value, label }: { value: string; label: string }) => { const [isVisible, setIsVisible] = useState(false) @@ -41,7 +39,7 @@ const MaskedUrl = ({ value, label }: { value: string; label: string }) => { size="tiny" aria-label={`${isVisible ? 'Hide' : 'Show'} ${label.replace(/^[A-Z][a-z]+/, (word) => word.toLowerCase())}`} aria-pressed={isVisible} - icon={isVisible ? : } + icon={isVisible ? : } onClick={() => setIsVisible(!isVisible)} /> @@ -93,7 +91,7 @@ export const PipelineDestinationReview = ({ 'newNamespaceName', ], }) - const rows: { label: string; value?: string; description?: string; isMasked?: boolean }[] = [] + const rows: { label: string; value?: string; description: string; isMasked?: boolean }[] = [] switch (type) { case 'BigQuery': rows.push( @@ -106,7 +104,7 @@ export const PipelineDestinationReview = ({ rows.push( { ...DUCKLAKE_CATALOG_URL_FIELD_COPY, - description: DUCKLAKE_CATALOG_URL_FIELD_COPY.createDescription, + description: 'Postgres URL of the database used for the DuckLake catalog.', value: catalogUrl, isMasked: true, }, @@ -177,11 +175,12 @@ export const PipelineDestinationReview = ({ )} ))} + diff --git a/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineReviewSummary.utils.ts b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineReviewSummary.utils.ts new file mode 100644 index 00000000000..a8f2b4233a6 --- /dev/null +++ b/apps/studio/components/interfaces/Database/Replication/CreatePipeline/PipelineReviewSummary.utils.ts @@ -0,0 +1,21 @@ +import type { DestinationPanelSchemaType } from '../DestinationPanel/DestinationForm/DestinationForm.schema' + +export const describeBigQueryTableLayout = ( + option: NonNullable[number] +) => { + const partition = option.partitionBy + let partitionDescription = 'No partitioning' + if (partition?.kind === 'time_column') { + partitionDescription = `Partition by ${partition.column || 'Not selected'} (${partition.granularity ?? 'day'})` + } + if (partition?.kind === 'ingestion_time') { + partitionDescription = `Partition by ingestion time (${partition.granularity ?? 'day'})` + } + if (partition?.kind === 'integer_range') { + partitionDescription = `Partition by ${partition.column || 'Not selected'} (range ${partition.start} to ${partition.end}, interval ${partition.interval})` + } + const clustering = option.clusterBy?.length + ? `Cluster by ${option.clusterBy.join(', ')}` + : 'No clustering' + return `${partitionDescription}; ${clustering}` +} diff --git a/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.test.tsx b/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.test.tsx index 23d9d27265f..d32c6663712 100644 --- a/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.test.tsx +++ b/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.test.tsx @@ -320,6 +320,7 @@ describe('DestinationForm edit submission', () => { expect(screen.queryByRole('region', { name: 'Destination review' }) !== null).toBe( shouldShowSection ) + expect(screen.queryByRole('region', { name: 'Data review' }) !== null).toBe(shouldShowSection) expect(screen.queryByRole('region', { name: 'Data' }) !== null).toBe(shouldShowSection) expect(screen.queryByRole('region', { name: 'Destination connection' }) !== null).toBe( shouldShowSection @@ -327,7 +328,7 @@ describe('DestinationForm edit submission', () => { if (shouldShowSection) { const section = screen.getByRole('region', { name: 'Destination connection' }) expect( - section.compareDocumentPosition(screen.getByText('Data')) & + section.compareDocumentPosition(screen.getByRole('region', { name: 'Data' })) & Node.DOCUMENT_POSITION_FOLLOWING ).toBeTruthy() } diff --git a/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.tsx b/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.tsx index 5b3a9aca81f..0ee4fe40469 100644 --- a/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.tsx +++ b/apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/index.tsx @@ -11,6 +11,7 @@ import { Admonition } from 'ui-patterns/Admonition' import * as z from 'zod' import { PipelineConnectionSection } from '../../CreatePipeline/PipelineConnectionSection' +import { PipelineDataReview } from '../../CreatePipeline/PipelineDataReview' import { PipelineDataSection } from '../../CreatePipeline/PipelineDataSection' import { PipelineDestinationReview } from '../../CreatePipeline/PipelineDestinationReview' import { @@ -542,6 +543,12 @@ export const DestinationForm = ({ <> + + )}