feat(studio): add support for new ducklake destination in replication UI (#45370)

This commit is contained in:
Coenen Benjamin authored and GitHub committed 2026-04-29 21:42:05 +02:00
1 parent 49ff97af06
commit 151a1792d9
18 files changed
+851 -32

No files matched your search

@@ -32,6 +32,18 @@ export const DestinationPanelFormSchema = z.object({
s3AccessKeyId: z.string().optional(),
s3SecretAccessKey: z.string().optional(),
s3Region: z.string().optional(),
// DuckLake fields
ducklakeCatalogUrl: z.string().optional(),
ducklakeDataPath: z.string().optional(),
ducklakePoolSize: z.number().int().min(1).max(6).optional(),
ducklakeS3AccessKeyId: z.string().optional(),
ducklakeS3SecretAccessKey: z.string().optional(),
ducklakeS3Region: z.string().optional(),
ducklakeS3Endpoint: z.string().optional(),
ducklakeS3UrlStyle: z.enum(['path', 'vhost']).optional(),
ducklakeS3UseSsl: z.boolean().optional(),
ducklakeMetadataSchema: z.string().optional(),
ducklakeExpireSnapshotsOlderThan: z.string().optional(),
})
export type DestinationPanelSchemaType = z.infer<typeof DestinationPanelFormSchema>
@@ -0,0 +1,166 @@
import { describe, expect, it, vi } from 'vitest'
import {
buildDestinationConfig,
buildDestinationConfigForValidation,
getDucklakeValidationIssues,
} from './DestinationForm.utils'
const baseDucklakeFormData = {
name: 'DuckLake Destination',
publicationName: 'pub',
maxFillMs: undefined,
maxTableSyncWorkers: undefined,
maxCopyConnectionsPerTable: undefined,
invalidatedSlotBehavior: undefined,
projectId: undefined,
datasetId: undefined,
serviceAccountKey: undefined,
connectionPoolSize: undefined,
maxStalenessMins: undefined,
warehouseName: undefined,
namespace: undefined,
newNamespaceName: undefined,
catalogToken: undefined,
s3AccessKeyId: undefined,
s3SecretAccessKey: undefined,
s3Region: undefined,
ducklakeCatalogUrl: 'postgres://user:pass@host:5432/catalog',
ducklakeDataPath: 's3://bucket/path',
ducklakePoolSize: 4,
ducklakeS3AccessKeyId: ' access-key ',
ducklakeS3SecretAccessKey: ' secret-key ',
ducklakeS3Region: ' eu-west-1 ',
ducklakeS3Endpoint: ' s3.example.com ',
ducklakeS3UrlStyle: 'path' as const,
ducklakeS3UseSsl: true,
ducklakeMetadataSchema: ' ducklake_metadata ',
ducklakeExpireSnapshotsOlderThan: ' 7 days ',
}
describe('DestinationForm.utils DuckLake', () => {
it('builds DuckLake validation config with required fields trimmed and blank optionals removed', () => {
const config = buildDestinationConfigForValidation({
projectRef: 'project-ref',
selectedType: 'DuckLake',
data: {
...baseDucklakeFormData,
ducklakeMetadataSchema: ' ',
ducklakeExpireSnapshotsOlderThan: ' ',
},
})
expect(config).toEqual({
ducklake: {
catalogUrl: 'postgres://user:pass@host:5432/catalog',
dataPath: 's3://bucket/path',
poolSize: 4,
s3AccessKeyId: 'access-key',
s3SecretAccessKey: 'secret-key',
s3Region: 'eu-west-1',
s3Endpoint: 's3.example.com',
s3UrlStyle: 'path',
s3UseSsl: true,
metadataSchema: undefined,
expireSnapshotsOlderThan: undefined,
},
})
})
it('builds DuckLake submit config with normalized values', async () => {
const createS3AccessKey = vi.fn()
const resolveNamespace = vi.fn()
const config = await buildDestinationConfig({
projectRef: 'project-ref',
selectedType: 'DuckLake',
data: baseDucklakeFormData,
createS3AccessKey,
resolveNamespace,
})
expect(config).toEqual({
ducklake: {
catalogUrl: 'postgres://user:pass@host:5432/catalog',
dataPath: 's3://bucket/path',
poolSize: 4,
s3AccessKeyId: 'access-key',
s3SecretAccessKey: 'secret-key',
s3Region: 'eu-west-1',
s3Endpoint: 's3.example.com',
s3UrlStyle: 'path',
s3UseSsl: true,
metadataSchema: 'ducklake_metadata',
expireSnapshotsOlderThan: '7 days',
},
})
expect(createS3AccessKey).not.toHaveBeenCalled()
expect(resolveNamespace).not.toHaveBeenCalled()
})
it('returns required-field errors for missing DuckLake settings', () => {
const issues = getDucklakeValidationIssues({
ducklakeCatalogUrl: '',
ducklakeDataPath: '',
ducklakeS3AccessKeyId: '',
ducklakeS3SecretAccessKey: '',
ducklakeS3Region: '',
ducklakeS3Endpoint: '',
ducklakeMetadataSchema: '',
})
expect(issues).toEqual([
{ path: 'ducklakeCatalogUrl', message: 'Catalog URL is required' },
{ path: 'ducklakeDataPath', message: 'Data path is required' },
{ path: 'ducklakeS3AccessKeyId', message: 'S3 Access Key ID is required' },
{ path: 'ducklakeS3SecretAccessKey', message: 'S3 Secret Access Key is required' },
{ path: 'ducklakeS3Region', message: 'S3 Region is required' },
{ path: 'ducklakeS3Endpoint', message: 'S3 Endpoint is required' },
])
})
it('returns format errors for invalid DuckLake values', () => {
const issues = getDucklakeValidationIssues({
ducklakeCatalogUrl: 'mysql://catalog',
ducklakeDataPath: 'file://bucket/path',
ducklakeS3AccessKeyId: 'access-key',
ducklakeS3SecretAccessKey: 'secret-key',
ducklakeS3Region: 'eu-west-1',
ducklakeS3Endpoint: 'https://s3.example.com',
ducklakeMetadataSchema: 'ducklake-schema',
})
expect(issues).toEqual([
{
path: 'ducklakeCatalogUrl',
message: 'DuckLake catalog URL must be a PostgreSQL-compatible URL',
},
{
path: 'ducklakeDataPath',
message: 'DuckLake data path must start with s3:// and cannot contain file://',
},
{
path: 'ducklakeS3Endpoint',
message: 'S3 endpoint should not contain the protocol scheme',
},
{
path: 'ducklakeMetadataSchema',
message: 'DuckLake metadata schema must contain only letters, numbers, and underscores',
},
])
})
it('accepts a valid DuckLake configuration', () => {
expect(
getDucklakeValidationIssues({
ducklakeCatalogUrl: 'postgresql://user:pass@host:5432/catalog',
ducklakeDataPath: 's3://bucket/path',
ducklakeS3AccessKeyId: 'access-key',
ducklakeS3SecretAccessKey: 'secret-key',
ducklakeS3Region: 'eu-west-1',
ducklakeS3Endpoint: 's3.example.com',
ducklakeMetadataSchema: 'ducklake_schema_1',
})
).toEqual([])
})
})
@@ -4,10 +4,14 @@ import z from 'zod'
import { DestinationType } from '../DestinationPanel.types'
import { CREATE_NEW_KEY, CREATE_NEW_NAMESPACE } from './DestinationForm.constants'
import { DestinationPanelFormSchema } from './DestinationForm.schema'
import {
DestinationPanelFormSchema,
type DestinationPanelSchemaType,
} from './DestinationForm.schema'
import {
BigQueryDestinationConfig,
DestinationConfig,
DucklakeDestinationConfig,
IcebergDestinationConfig,
} from '@/data/replication/create-destination-pipeline-mutation'
import {
@@ -16,6 +20,102 @@ import {
} from '@/data/storage/s3-access-key-create-mutation'
import { ResponseError } from '@/types'
const normalizeOptionalString = (value?: string) => {
const trimmed = value?.trim()
return trimmed ? trimmed : undefined
}
const normalizeRequiredString = (value?: string) => value?.trim() ?? ''
type DucklakeFieldPath =
| 'ducklakeCatalogUrl'
| 'ducklakeDataPath'
| 'ducklakeS3AccessKeyId'
| 'ducklakeS3SecretAccessKey'
| 'ducklakeS3Region'
| 'ducklakeS3Endpoint'
| 'ducklakeMetadataSchema'
export type DucklakeValidationIssue = {
path: DucklakeFieldPath
message: string
}
export const getDucklakeValidationIssues = (
data: Pick<
DestinationPanelSchemaType,
| 'ducklakeCatalogUrl'
| 'ducklakeDataPath'
| 'ducklakeS3AccessKeyId'
| 'ducklakeS3SecretAccessKey'
| 'ducklakeS3Region'
| 'ducklakeS3Endpoint'
| 'ducklakeMetadataSchema'
>
): DucklakeValidationIssue[] => {
const issues: DucklakeValidationIssue[] = []
if (!data.ducklakeCatalogUrl?.length) {
issues.push({ path: 'ducklakeCatalogUrl', message: 'Catalog URL is required' })
} else if (
!data.ducklakeCatalogUrl.startsWith('postgres://') &&
!data.ducklakeCatalogUrl.startsWith('postgresql://')
) {
issues.push({
path: 'ducklakeCatalogUrl',
message: 'DuckLake catalog URL must be a PostgreSQL-compatible URL',
})
}
if (!data.ducklakeDataPath?.length) {
issues.push({ path: 'ducklakeDataPath', message: 'Data path is required' })
} else if (
!data.ducklakeDataPath.startsWith('s3://') ||
data.ducklakeDataPath.includes('file://')
) {
issues.push({
path: 'ducklakeDataPath',
message: 'DuckLake data path must start with s3:// and cannot contain file://',
})
}
if (!data.ducklakeS3AccessKeyId?.length) {
issues.push({ path: 'ducklakeS3AccessKeyId', message: 'S3 Access Key ID is required' })
}
if (!data.ducklakeS3SecretAccessKey?.length) {
issues.push({
path: 'ducklakeS3SecretAccessKey',
message: 'S3 Secret Access Key is required',
})
}
if (!data.ducklakeS3Region?.length) {
issues.push({ path: 'ducklakeS3Region', message: 'S3 Region is required' })
}
if (!data.ducklakeS3Endpoint?.length) {
issues.push({ path: 'ducklakeS3Endpoint', message: 'S3 Endpoint is required' })
} else if (
data.ducklakeS3Endpoint.startsWith('http://') ||
data.ducklakeS3Endpoint.startsWith('https://')
) {
issues.push({
path: 'ducklakeS3Endpoint',
message: 'S3 endpoint should not contain the protocol scheme',
})
}
if (data.ducklakeMetadataSchema && !/^[A-Za-z0-9_]+$/.test(data.ducklakeMetadataSchema)) {
issues.push({
path: 'ducklakeMetadataSchema',
message: 'DuckLake metadata schema must contain only letters, numbers, and underscores',
})
}
return issues
}
// Helper function to build destination config for validation
export const buildDestinationConfigForValidation = ({
projectRef,
@@ -63,6 +163,22 @@ export const buildDestinationConfigForValidation = ({
s3Region: data.s3Region ?? '',
},
}
} else if (selectedType === 'DuckLake') {
return {
ducklake: {
catalogUrl: data.ducklakeCatalogUrl ?? '',
dataPath: data.ducklakeDataPath ?? '',
poolSize: data.ducklakePoolSize,
s3AccessKeyId: normalizeRequiredString(data.ducklakeS3AccessKeyId),
s3SecretAccessKey: normalizeRequiredString(data.ducklakeS3SecretAccessKey),
s3Region: normalizeRequiredString(data.ducklakeS3Region),
s3Endpoint: normalizeRequiredString(data.ducklakeS3Endpoint),
s3UrlStyle: data.ducklakeS3UrlStyle,
s3UseSsl: data.ducklakeS3UseSsl,
metadataSchema: normalizeOptionalString(data.ducklakeMetadataSchema),
expireSnapshotsOlderThan: normalizeOptionalString(data.ducklakeExpireSnapshotsOlderThan),
},
}
} else {
throw new Error('Invalid destination type')
}
@@ -127,6 +243,21 @@ export const buildDestinationConfig = async ({
s3Region: data.s3Region ?? '',
}
destinationConfig = { iceberg: icebergConfig }
} else if (selectedType === 'DuckLake') {
const ducklakeConfig: DucklakeDestinationConfig = {
catalogUrl: data.ducklakeCatalogUrl ?? '',
dataPath: data.ducklakeDataPath ?? '',
poolSize: data.ducklakePoolSize,
s3AccessKeyId: normalizeRequiredString(data.ducklakeS3AccessKeyId),
s3SecretAccessKey: normalizeRequiredString(data.ducklakeS3SecretAccessKey),
s3Region: normalizeRequiredString(data.ducklakeS3Region),
s3Endpoint: normalizeRequiredString(data.ducklakeS3Endpoint),
s3UrlStyle: data.ducklakeS3UrlStyle,
s3UseSsl: data.ducklakeS3UseSsl,
metadataSchema: normalizeOptionalString(data.ducklakeMetadataSchema),
expireSnapshotsOlderThan: normalizeOptionalString(data.ducklakeExpireSnapshotsOlderThan),
}
destinationConfig = { ducklake: ducklakeConfig }
}
return destinationConfig
@@ -93,6 +93,284 @@ export const BigQueryFields = ({ form }: { form: UseFormReturn<DestinationPanelS
)
}
export const DuckLakeFields = ({ form }: { form: UseFormReturn<DestinationPanelSchemaType> }) => {
const [showCatalogUrl, setShowCatalogUrl] = useState(false)
const [showSecretAccessKey, setShowSecretAccessKey] = useState(false)
return (
<div className="flex flex-col gap-y-6 p-5">
<p className="text-sm font-medium text-foreground">DuckLake settings</p>
<div className="flex flex-col gap-y-1">
<p className="text-sm font-medium text-foreground">Catalog</p>
<p className="text-sm text-foreground-light">
Configure the PostgreSQL-backed DuckLake catalog and the S3-compatible storage location
for replicated data.
</p>
</div>
<div className="flex flex-col gap-y-4">
<FormField
control={form.control}
name="ducklakeCatalogUrl"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Catalog URL"
description="A PostgreSQL connection string for the DuckLake catalog"
>
<FormControl>
<Input
value={field.value ?? ''}
type={showCatalogUrl ? 'text' : 'password'}
placeholder="postgres://user:pass@host:5432/ducklake_catalog"
onChange={(event) => field.onChange(event.target.value)}
actions={
<div className="flex items-center justify-center">
<Button
type="default"
className="w-7"
icon={showCatalogUrl ? <Eye /> : <EyeOff />}
onClick={() => setShowCatalogUrl(!showCatalogUrl)}
/>
</div>
}
/>
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeDataPath"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Data path"
description="An S3 path where DuckLake data files will be written"
>
<FormControl>
<Input_Shadcn_
{...field}
placeholder="s3://bucket/path"
value={field.value ?? ''}
/>
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakePoolSize"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Pool size"
description="Optional number of concurrent DuckDB connections to use"
>
<FormControl>
<Input_Shadcn_
type="number"
min={1}
max={6}
value={field.value ?? ''}
placeholder="Default: 4"
onChange={(event) =>
field.onChange(
event.target.value === '' ? undefined : Number(event.target.value)
)
}
/>
</FormControl>
</FormItemLayout>
)}
/>
</div>
<div className="flex flex-col gap-y-1">
<p className="text-sm font-medium text-foreground">Object storage</p>
<p className="text-sm text-foreground-light">
Optional credentials and endpoint settings for S3-compatible storage providers.
</p>
</div>
<div className="flex flex-col gap-y-4">
<FormField
control={form.control}
name="ducklakeS3AccessKeyId"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="S3 Access Key ID"
description="Required access key ID for the object storage provider"
>
<FormControl>
<Input_Shadcn_ {...field} placeholder="my-access-key" value={field.value ?? ''} />
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeS3SecretAccessKey"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="S3 Secret Access Key"
description="Required secret access key for the object storage provider"
className="relative"
>
<FormControl>
<Input_Shadcn_
{...field}
type={showSecretAccessKey ? 'text' : 'password'}
placeholder="my-secret-key"
value={field.value ?? ''}
/>
</FormControl>
<Button
type="default"
icon={showSecretAccessKey ? <Eye /> : <EyeOff />}
className="w-7 absolute right-6 top-[4px]"
onClick={() => setShowSecretAccessKey(!showSecretAccessKey)}
/>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeS3Region"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="S3 Region"
description="Required region for the object storage provider"
>
<FormControl>
<Input_Shadcn_ {...field} placeholder="us-east-1" value={field.value ?? ''} />
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeS3Endpoint"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="S3 Endpoint"
description="Required endpoint without the protocol scheme, for example `127.0.0.1:5000/s3`"
>
<FormControl>
<Input_Shadcn_
{...field}
placeholder="127.0.0.1:5000/s3"
value={field.value ?? ''}
/>
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeS3UrlStyle"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="S3 URL style"
description="Choose `path` for MinIO/Supabase-style endpoints or `vhost` for AWS-style virtual host addressing"
>
<FormControl>
<Select_Shadcn_ value={field.value ?? 'path'} onValueChange={field.onChange}>
<SelectTrigger_Shadcn_>{field.value ?? 'path'}</SelectTrigger_Shadcn_>
<SelectContent_Shadcn_>
<SelectItem_Shadcn_ value="path">path</SelectItem_Shadcn_>
<SelectItem_Shadcn_ value="vhost">vhost</SelectItem_Shadcn_>
</SelectContent_Shadcn_>
</Select_Shadcn_>
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeS3UseSsl"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Use SSL"
description="Whether to use SSL when connecting to the S3-compatible endpoint"
>
<FormControl>
<Select_Shadcn_
value={field.value === false ? 'false' : 'true'}
onValueChange={(value) => field.onChange(value === 'true')}
>
<SelectTrigger_Shadcn_>
{field.value === false ? 'false' : 'true'}
</SelectTrigger_Shadcn_>
<SelectContent_Shadcn_>
<SelectItem_Shadcn_ value="true">true</SelectItem_Shadcn_>
<SelectItem_Shadcn_ value="false">false</SelectItem_Shadcn_>
</SelectContent_Shadcn_>
</Select_Shadcn_>
</FormControl>
</FormItemLayout>
)}
/>
</div>
<div className="flex flex-col gap-y-1">
<p className="text-sm font-medium text-foreground">Maintenance</p>
<p className="text-sm text-foreground-light">
Optional settings for DuckLake metadata tables and snapshot cleanup.
</p>
</div>
<div className="flex flex-col gap-y-4">
<FormField
control={form.control}
name="ducklakeMetadataSchema"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Metadata schema"
description="Schema used for DuckLake metadata tables in PostgreSQL"
>
<FormControl>
<Input_Shadcn_ {...field} placeholder="ducklake" value={field.value ?? ''} />
</FormControl>
</FormItemLayout>
)}
/>
<FormField
control={form.control}
name="ducklakeExpireSnapshotsOlderThan"
render={({ field }) => (
<FormItemLayout
layout="horizontal"
label="Expire snapshots older than"
description="Optional snapshot retention interval, for example `7 days`"
>
<FormControl>
<Input_Shadcn_ {...field} placeholder="7 days" value={field.value ?? ''} />
</FormControl>
</FormItemLayout>
)}
/>
</div>
</div>
)
}
/**
* [Joshen] JFYI I'd foresee a possible UX friction point here regarding S3 access key IDs and secret access keys
* - We'd allow users to select access key IDs via a dropdown here, but require a text input for secret access keys
@@ -11,6 +11,7 @@ import * as z from 'zod'
import {
useIsETLBigQueryPrivateAlpha,
useIsETLDucklakePrivateAlpha,
useIsETLIcebergPrivateAlpha,
} from '../../useIsETLPrivateAlpha'
import { DestinationType } from '../DestinationPanel.types'
@@ -20,9 +21,10 @@ import { DestinationPanelFormSchema as FormSchema } from './DestinationForm.sche
import {
buildDestinationConfig,
buildDestinationConfigForValidation,
getDucklakeValidationIssues,
} from './DestinationForm.utils'
import { DestinationNameInput } from './DestinationNameInput'
import { AnalyticsBucketFields, BigQueryFields } from './DestinationPanelFields'
import { AnalyticsBucketFields, BigQueryFields, DuckLakeFields } from './DestinationPanelFields'
import { NewPublicationPanel } from './NewPublicationPanel'
import { NoDestinationsAvailable } from './NoDestinationsAvailable'
import { PublicationSelection } from './PublicationSelection'
@@ -70,6 +72,20 @@ interface DestinationFormProps {
onClose: () => void
}
type DucklakeApiConfig = {
catalog_url: string
data_path: string
pool_size?: number
s3_access_key_id?: string
s3_secret_access_key?: string
s3_region?: string
s3_endpoint?: string
s3_url_style?: 'path' | 'vhost'
s3_use_ssl?: boolean
metadata_schema?: string
expire_snapshots_older_than?: string
}
export const DestinationForm = ({
selectedType,
visible,
@@ -81,6 +97,7 @@ export const DestinationForm = ({
const etlEnableBigQuery = useIsETLBigQueryPrivateAlpha()
const etlEnableIceberg = useIsETLIcebergPrivateAlpha()
const etlEnableDucklake = useIsETLDucklakePrivateAlpha()
const { can: canReadAPIKeys } = useAsyncCheckPermissions(PermissionAction.SECRETS_READ, '*')
const [isFormInteracting, setIsFormInteracting] = useState(false)
@@ -108,8 +125,9 @@ export const DestinationForm = ({
if (etlEnableBigQuery) destinations.push({ value: 'BigQuery', label: 'BigQuery' })
if (etlEnableIceberg)
destinations.push({ value: 'Analytics Bucket', label: 'Analytics Bucket' })
if (etlEnableDucklake) destinations.push({ value: 'DuckLake', label: 'DuckLake' })
return destinations
}, [etlEnableBigQuery, etlEnableIceberg])
}, [etlEnableBigQuery, etlEnableDucklake, etlEnableIceberg])
const hasNoAvailableDestinations = availableDestinations.length === 0
const { data: sourcesData } = useReplicationSourcesQuery({ projectRef })
@@ -171,6 +189,14 @@ export const DestinationForm = ({
const config = destinationData?.config
const isBigQueryConfig = config && 'big_query' in config
const isIcebergConfig = config && 'iceberg' in config
const ducklakeConfigValue =
config && 'ducklake' in (config as Record<string, unknown>)
? (config as Record<string, unknown>).ducklake
: undefined
const ducklakeConfig =
ducklakeConfigValue && typeof ducklakeConfigValue === 'object'
? (ducklakeConfigValue as DucklakeApiConfig)
: undefined
return {
// Common fields
@@ -199,6 +225,18 @@ export const DestinationForm = ({
s3SecretAccessKey: isIcebergConfig ? config.iceberg.supabase.s3_secret_access_key : '',
s3Region:
projectSettings?.region ?? (isIcebergConfig ? config.iceberg.supabase.s3_region : ''),
// DuckLake fields
ducklakeCatalogUrl: ducklakeConfig?.catalog_url ?? '',
ducklakeDataPath: ducklakeConfig?.data_path ?? '',
ducklakePoolSize: ducklakeConfig?.pool_size,
ducklakeS3AccessKeyId: ducklakeConfig?.s3_access_key_id ?? '',
ducklakeS3SecretAccessKey: ducklakeConfig?.s3_secret_access_key ?? '',
ducklakeS3Region: ducklakeConfig?.s3_region ?? '',
ducklakeS3Endpoint: ducklakeConfig?.s3_endpoint ?? '',
ducklakeS3UrlStyle: ducklakeConfig?.s3_url_style ?? 'path',
ducklakeS3UseSsl: ducklakeConfig?.s3_use_ssl ?? true,
ducklakeMetadataSchema: ducklakeConfig?.metadata_schema ?? 'ducklake',
ducklakeExpireSnapshotsOlderThan: ducklakeConfig?.expire_snapshots_older_than ?? '',
}
}, [destinationData, pipelineData, catalogToken, projectSettings])
@@ -244,6 +282,10 @@ export const DestinationForm = ({
if (data.s3AccessKeyId !== 'create-new' && !data.s3SecretAccessKey?.length) {
addRequiredFieldError('s3SecretAccessKey', 'S3 Secret Access Key is required')
}
} else if (selectedType === 'DuckLake') {
getDucklakeValidationIssues(data).forEach(({ path, message }) => {
addRequiredFieldError(path, message)
})
}
})
),
@@ -536,6 +578,8 @@ export const DestinationForm = ({
setIsFormInteracting={setIsFormInteracting}
onSelectNewBucket={() => setNewBucketSheetVisible(true)}
/>
) : selectedType === 'DuckLake' && etlEnableDucklake ? (
<DuckLakeFields form={form} />
) : null}
<DialogSectionSeparator />
@@ -41,6 +41,7 @@ export const DestinationPanel = ({ onSuccessCreateReadReplica }: DestinationPane
'Read Replica',
'BigQuery',
'Analytics Bucket',
'DuckLake',
]).withOptions({
history: 'push',
clearOnDefault: true,
@@ -1 +1 @@
export type DestinationType = 'Read Replica' | 'BigQuery' | 'Analytics Bucket'
export type DestinationType = 'Read Replica' | 'BigQuery' | 'Analytics Bucket' | 'DuckLake'
@@ -3,7 +3,11 @@ import { parseAsInteger, parseAsStringEnum, useQueryState } from 'nuqs'
import { Badge, cn, RadioGroupStacked, RadioGroupStackedItem } from 'ui'
import { useDestinationInformation } from '../useDestinationInformation'
import { useIsETLBigQueryPrivateAlpha, useIsETLIcebergPrivateAlpha } from '../useIsETLPrivateAlpha'
import {
useIsETLBigQueryPrivateAlpha,
useIsETLDucklakePrivateAlpha,
useIsETLIcebergPrivateAlpha,
} from '../useIsETLPrivateAlpha'
import { DestinationType } from './DestinationPanel.types'
import { InlineLink } from '@/components/ui/InlineLink'
import { useIsFeatureEnabled } from '@/hooks/misc/useIsFeatureEnabled'
@@ -11,11 +15,15 @@ import { useIsFeatureEnabled } from '@/hooks/misc/useIsFeatureEnabled'
export const DestinationTypeSelection = () => {
const etlEnableBigQuery = useIsETLBigQueryPrivateAlpha()
const etlEnableIceberg = useIsETLIcebergPrivateAlpha()
const etlEnableDucklake = useIsETLDucklakePrivateAlpha()
const { infrastructureReadReplicas } = useIsFeatureEnabled(['infrastructure:read_replicas'])
const numberOfTypes = [infrastructureReadReplicas, etlEnableBigQuery, etlEnableIceberg].filter(
Boolean
).length
const numberOfTypes = [
infrastructureReadReplicas,
etlEnableBigQuery,
etlEnableIceberg,
etlEnableDucklake,
].filter(Boolean).length
const [urlDestinationType, setDestinationType] = useQueryState(
'destinationType',
@@ -23,6 +31,7 @@ export const DestinationTypeSelection = () => {
'Read Replica',
'BigQuery',
'Analytics Bucket',
'DuckLake',
]).withOptions({
history: 'push',
clearOnDefault: true,
@@ -51,8 +60,12 @@ export const DestinationTypeSelection = () => {
value={destinationType}
onValueChange={(value) => setDestinationType(value as DestinationType)}
className={cn(
'grid [&>button>div]:py-4 grid-cols-3',
numberOfTypes === 3 && !editMode ? 'grid-cols-3' : 'grid-cols-2',
'grid [&>button>div]:py-4',
!editMode && numberOfTypes >= 4
? 'grid-cols-4'
: !editMode && numberOfTypes === 3
? 'grid-cols-3'
: 'grid-cols-2',
'[&>button:first-of-type]:rounded-none [&>button:last-of-type]:rounded-none',
'[&>button:first-of-type]:!rounded-l-lg [&>button:last-of-type]:!rounded-r-lg'
)}
@@ -118,6 +131,24 @@ export const DestinationTypeSelection = () => {
</div>
</RadioGroupStackedItem>
)}
{((!editMode && etlEnableDucklake) || (editMode && destinationType === 'DuckLake')) && (
<RadioGroupStackedItem label="" showIndicator={false} id="DuckLake" value="DuckLake">
<div className="flex flex-col gap-y-2">
<Database size={20} />
<div className="flex flex-col gap-y-0.5 text-sm text-left">
<div className="flex items-center gap-x-2">
<p>DuckLake</p>
<Badge>Alpha</Badge>
</div>
<p className="text-foreground-lighter">
Send data to a DuckLake catalog backed by S3-compatible object storage for
flexible lakehouse workflows
</p>
</div>
</div>
</RadioGroupStackedItem>
)}
</RadioGroupStacked>
{destinationType !== 'Read Replica' && (
@@ -138,6 +138,8 @@ export const DestinationRow = ({ destinationId }: DestinationRowProps) => {
<BigQuery size={18} className="text-foreground-light" />
) : type === 'Analytics Bucket' ? (
<AnalyticsBucket size={18} className="text-foreground-light" />
) : type === 'DuckLake' ? (
<Database size={18} className="text-foreground-light" />
) : (
<Database size={18} className="text-foreground-light" />
)}
@@ -29,7 +29,11 @@ import { DestinationRow } from './DestinationRow'
import { DisableExternalReplicationDialog } from './DisableExternalReplicationDialog'
import { PIPELINE_ERROR_MESSAGES } from './Pipeline.utils'
import { ReadReplicaRow } from './ReadReplicas/ReadReplicaRow'
import { useIsETLBigQueryPrivateAlpha, useIsETLIcebergPrivateAlpha } from './useIsETLPrivateAlpha'
import {
useIsETLBigQueryPrivateAlpha,
useIsETLDucklakePrivateAlpha,
useIsETLIcebergPrivateAlpha,
} from './useIsETLPrivateAlpha'
import { AlertError } from '@/components/ui/AlertError'
import { DocsButton } from '@/components/ui/DocsButton'
import { useReadReplicasQuery } from '@/data/read-replicas/replicas-query'
@@ -47,6 +51,7 @@ export const Destinations = () => {
const etlEnableBigQuery = useIsETLBigQueryPrivateAlpha()
const etlEnableIceberg = useIsETLIcebergPrivateAlpha()
const etlEnableDucklake = useIsETLDucklakePrivateAlpha()
const { infrastructureReadReplicas } = useIsFeatureEnabled(['infrastructure:read_replicas'])
const newDestinationDefaultType = infrastructureReadReplicas
@@ -55,7 +60,9 @@ export const Destinations = () => {
? 'BigQuery'
: etlEnableIceberg
? 'Analytics Bucket'
: null
: etlEnableDucklake
? 'DuckLake'
: null
const prefetchedRef = useRef(false)
const [filterString, setFilterString] = useState<string>('')
@@ -69,6 +76,7 @@ export const Destinations = () => {
'Read Replica',
'BigQuery',
'Analytics Bucket',
'DuckLake',
]).withOptions({
history: 'push',
clearOnDefault: true,
@@ -1,13 +1,14 @@
import { Handle, Position } from '@xyflow/react'
import { useParams } from 'common'
import { AnalyticsBucket, BigQuery, Database } from 'icons'
import { PropsWithChildren, useMemo } from 'react'
import { ComponentType, PropsWithChildren, useMemo } from 'react'
import { AWS_REGIONS } from 'shared-data'
import { cn, Tooltip, TooltipContent, TooltipTrigger } from 'ui'
import { getStatusName } from '../Pipeline.utils'
import { getStatusLabel } from '../ReadReplicas/ReadReplicas.utils'
import { STATUS_REFRESH_FREQUENCY_MS } from '../Replication.constants'
import { getReplicationDestinationType, type ReplicationDestinationType } from './Nodes.utils'
import { useReadReplicasQuery } from '@/data/read-replicas/replicas-query'
import { formatDatabaseID } from '@/data/read-replicas/replicas.utils'
import { useReplicationDestinationsQuery } from '@/data/replication/destinations-query'
@@ -18,6 +19,15 @@ import { BASE_PATH } from '@/lib/constants'
export const NODE_WIDTH = 480
const destinationIconByType: Record<
ReplicationDestinationType,
ComponentType<{ className?: string; size?: string | number }>
> = {
BigQuery,
'Analytics Bucket': AnalyticsBucket,
DuckLake: Database,
}
const NodeContainer = ({ className, children }: PropsWithChildren<{ className?: string }>) => {
return (
<div
@@ -84,17 +94,12 @@ export const ReplicationNode = ({ id }: { id: string }) => {
)
const statusName = getStatusName(pipelineStatusData?.status)
const config = destination?.config ?? {}
const type =
'big_query' in config ? 'BigQuery' : 'iceberg' in config ? 'Analytics Bucket' : undefined
const type = getReplicationDestinationType(destination?.config)
const DestinationIcon = type ? destinationIconByType[type] : undefined
return (
<NodeContainer className="justify-start gap-x-3">
{type === 'BigQuery' ? (
<BigQuery size={20} className="text-foreground-light" />
) : type === 'Analytics Bucket' ? (
<AnalyticsBucket size={20} className="text-foreground-light" />
) : null}
{DestinationIcon ? <DestinationIcon size={20} className="text-foreground-light" /> : null}
<div className="text-sm flex flex-col gap-y-0.5">
<div className="flex items-center">
<p>{type}</p>
@@ -0,0 +1,22 @@
import { describe, expect, it } from 'vitest'
import { getReplicationDestinationType } from './Nodes.utils'
describe('getReplicationDestinationType', () => {
it('returns BigQuery for big_query configs', () => {
expect(getReplicationDestinationType({ big_query: {} })).toBe('BigQuery')
})
it('returns Analytics Bucket for iceberg configs', () => {
expect(getReplicationDestinationType({ iceberg: {} })).toBe('Analytics Bucket')
})
it('returns DuckLake for ducklake configs', () => {
expect(getReplicationDestinationType({ ducklake: {} })).toBe('DuckLake')
})
it('returns undefined for unknown or missing configs', () => {
expect(getReplicationDestinationType({})).toBeUndefined()
expect(getReplicationDestinationType(undefined)).toBeUndefined()
})
})
@@ -0,0 +1,11 @@
export type ReplicationDestinationType = 'BigQuery' | 'Analytics Bucket' | 'DuckLake'
export const getReplicationDestinationType = (
config?: Record<string, unknown>
): ReplicationDestinationType | undefined => {
if (!config) return undefined
if ('big_query' in config) return 'BigQuery'
if ('iceberg' in config) return 'Analytics Bucket'
if ('ducklake' in config) return 'DuckLake'
return undefined
}
@@ -2,6 +2,7 @@ import { useParams } from 'common'
import { DestinationType } from './DestinationPanel/DestinationPanel.types'
import { getStatusName } from './Pipeline.utils'
import { getReplicationDestinationType } from './ReplicationDiagram/Nodes.utils'
import { useReplicationDestinationByIdQuery } from '@/data/replication/destination-by-id-query'
import { useReplicationPipelineStatusQuery } from '@/data/replication/pipeline-status-query'
import { useReplicationPipelinesQuery } from '@/data/replication/pipelines-query'
@@ -26,13 +27,9 @@ export const useDestinationInformation = ({ id }: { id?: number | null }) => {
projectRef,
destinationId: id,
})
const destinationType: DestinationType | undefined = !destination
? undefined
: 'big_query' in destination.config
? 'BigQuery'
: 'iceberg' in destination.config
? 'Analytics Bucket'
: undefined
const destinationType: DestinationType | undefined = getReplicationDestinationType(
destination?.config as Record<string, unknown> | undefined
)
const {
data: pipelinesData,
@@ -31,9 +31,14 @@ export const useIsETLIcebergPrivateAlpha = () => {
return useIsCurrentOrgInFlagList('etlEnableIcebergPrivateAlpha')
}
export const useIsETLDucklakePrivateAlpha = () => {
return useIsCurrentOrgInFlagList('etlEnableDucklakePrivateAlpha')
}
export const useIsETLPrivateAlpha = () => {
const hasAccessToETLBigQuery = useIsCurrentOrgInFlagList('etlEnableBigQueryPrivateAlpha')
const hasAccessToETLIceberg = useIsCurrentOrgInFlagList('etlEnableIcebergPrivateAlpha')
const hasAccessToETLDucklake = useIsCurrentOrgInFlagList('etlEnableDucklakePrivateAlpha')
return hasAccessToETLBigQuery || hasAccessToETLIceberg
return hasAccessToETLBigQuery || hasAccessToETLIceberg || hasAccessToETLDucklake
}
@@ -13,6 +13,9 @@ export type DestinationConfig =
| {
iceberg: IcebergDestinationConfig
}
| {
ducklake: DucklakeDestinationConfig
}
export type BigQueryDestinationConfig = {
projectId: string
@@ -32,6 +35,20 @@ export type IcebergDestinationConfig = {
s3Region: string
}
export type DucklakeDestinationConfig = {
catalogUrl: string
dataPath: string
poolSize?: number
s3AccessKeyId: string
s3SecretAccessKey: string
s3Region: string
s3Endpoint: string
s3UrlStyle?: 'path' | 'vhost'
s3UseSsl?: boolean
metadataSchema?: string
expireSnapshotsOlderThan?: string
}
export type BatchConfig = {
maxFillMs?: number
}
@@ -108,8 +125,38 @@ async function createDestinationPipeline(
},
},
}
} else if ('ducklake' in destinationConfig) {
const {
catalogUrl,
dataPath,
poolSize,
s3AccessKeyId,
s3SecretAccessKey,
s3Region,
s3Endpoint,
s3UrlStyle,
s3UseSsl,
metadataSchema,
expireSnapshotsOlderThan,
} = destinationConfig.ducklake
destination_config = {
ducklake: {
catalog_url: catalogUrl,
data_path: dataPath,
pool_size: poolSize,
s3_access_key_id: s3AccessKeyId,
s3_secret_access_key: s3SecretAccessKey,
s3_region: s3Region,
s3_endpoint: s3Endpoint,
s3_url_style: s3UrlStyle,
s3_use_ssl: s3UseSsl,
metadata_schema: metadataSchema,
expire_snapshots_older_than: expireSnapshotsOlderThan,
},
} as unknown as components['schemas']['CreateReplicationDestinationPipelineBody']['destination_config']
} else {
throw new Error('Invalid destination config: must specify either bigQuery or iceberg')
throw new Error('Invalid destination config: must specify bigQuery, iceberg, or ducklake')
}
const pipeline_config = {
@@ -81,8 +81,37 @@ async function updateDestinationPipeline(
},
},
}
} else if ('ducklake' in destinationConfig) {
const {
catalogUrl,
dataPath,
poolSize,
s3AccessKeyId,
s3SecretAccessKey,
s3Region,
s3Endpoint,
s3UrlStyle,
s3UseSsl,
metadataSchema,
expireSnapshotsOlderThan,
} = destinationConfig.ducklake
destination_config = {
ducklake: {
catalog_url: catalogUrl,
data_path: dataPath,
pool_size: poolSize,
s3_access_key_id: s3AccessKeyId,
s3_secret_access_key: s3SecretAccessKey,
s3_region: s3Region,
s3_endpoint: s3Endpoint,
s3_url_style: s3UrlStyle,
s3_use_ssl: s3UseSsl,
metadata_schema: metadataSchema,
expire_snapshots_older_than: expireSnapshotsOlderThan,
},
} as unknown as components['schemas']['UpdateReplicationDestinationPipelineBody']['destination_config']
} else {
throw new Error('Invalid destination config: must specify either bigQuery or iceberg')
throw new Error('Invalid destination config: must specify bigQuery, iceberg, or ducklake')
}
const pipeline_config = {
@@ -59,8 +59,38 @@ async function validateDestination(
},
},
}
} else if ('ducklake' in destinationConfig) {
const {
catalogUrl,
dataPath,
poolSize,
s3AccessKeyId,
s3SecretAccessKey,
s3Region,
s3Endpoint,
s3UrlStyle,
s3UseSsl,
metadataSchema,
expireSnapshotsOlderThan,
} = destinationConfig.ducklake
config = {
ducklake: {
catalog_url: catalogUrl,
data_path: dataPath,
pool_size: poolSize,
s3_access_key_id: s3AccessKeyId,
s3_secret_access_key: s3SecretAccessKey,
s3_region: s3Region,
s3_endpoint: s3Endpoint,
s3_url_style: s3UrlStyle,
s3_use_ssl: s3UseSsl,
metadata_schema: metadataSchema,
expire_snapshots_older_than: expireSnapshotsOlderThan,
},
} as unknown as components['schemas']['ValidateReplicationDestinationBody']['config']
} else {
throw new Error('Invalid destination config: must specify either bigQuery or iceberg')
throw new Error('Invalid destination config: must specify bigQuery, iceberg, or ducklake')
}
const { data, error } = await post('/platform/replication/{ref}/destinations/validate', {