diff --git a/.docker/docker-compose-infra.yml b/.docker/docker-compose-infra.yml index 056a84951..8cc153c0c 100644 --- a/.docker/docker-compose-infra.yml +++ b/.docker/docker-compose-infra.yml @@ -159,8 +159,11 @@ services: entrypoint: > /bin/sh -c " set -e; - for b in supa-storage-bucket warehouse--table-s3 supa-storage-profiles; do + for b in supa-storage-bucket warehouse--table-s3 warehouse-shard-2--table-s3 supa-storage-profiles; do curl -fsS --aws-sigv4 'aws:amz:us-east-1:s3' -u supa-storage:secret1234 -X PUT http://rustfs:9000/\$$b; + done; + for b in warehouse--table-s3 warehouse-shard-2--table-s3; do + curl -fsS --aws-sigv4 'aws:amz:us-east-1:s3' -u supa-storage:secret1234 -X PUT http://rustfs:9000/iceberg/v1/buckets/\$$b; done " @@ -189,22 +192,6 @@ services: - IMGPROXY_PRESETS=default=width:3000/height:8192 - IMGPROXY_FORMAT_QUALITY=jpeg=80,avif=62,webp=80 - rest-catalog: - image: apache/iceberg-rest-fixture:1.10.1 - container_name: iceberg-rest - depends_on: - rustfs_setup: - condition: service_completed_successfully - ports: - - 8181:8181 - environment: - - AWS_ACCESS_KEY_ID=supa-storage - - AWS_SECRET_ACCESS_KEY=secret1234 - - AWS_REGION=us-east-1 - - CATALOG_WAREHOUSE=s3://warehouse--table-s3/ - - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO - - CATALOG_S3_ENDPOINT=http://rustfs:9000 - # Optional for rate-limiting # redis: # image: redis:6.2-alpine diff --git a/.env.test.sample b/.env.test.sample index efa319860..1832927c4 100644 --- a/.env.test.sample +++ b/.env.test.sample @@ -28,11 +28,12 @@ REQUEST_X_FORWARDED_HOST_REGEXP= VECTOR_ENABLED=true VECTOR_S3_BUCKETS=supa-test-local-dev ICEBERG_ENABLED=true -ICEBERG_WAREHOUSE=. -ICEBERG_SHARDS=. -ICEBERG_CATALOG_URL=http://127.0.0.1:8181/v1 -ICEBERG_CATALOG_AUTH_TYPE=token -ICEBERG_CATALOG_AUTH_TOKEN=token +ICEBERG_WAREHOUSE=warehouse--table-s3 +ICEBERG_SHARDS=warehouse--table-s3,warehouse-shard-2--table-s3 +ICEBERG_CATALOG_URL=http://127.0.0.1:9000/iceberg/v1 +ICEBERG_CATALOG_AUTH_TYPE=sigv4 +ICEBERG_CATALOG_SIGV4_SERVICE=s3 +ICEBERG_CATALOG_PURGE_ON_DROP=false ICEBERG_S3_DELETE_ENABLED=true ICEBERG_BUCKET_DETECTION_MODE="BUCKET" diff --git a/acceptance/specs/iceberg.test.ts b/acceptance/specs/iceberg.test.ts index 3f3767a0e..96ae932b5 100644 --- a/acceptance/specs/iceberg.test.ts +++ b/acceptance/specs/iceberg.test.ts @@ -447,7 +447,7 @@ function createTablePayload(tableName: string) { 'schema-id': 0, type: 'struct', }, - spec: { + 'partition-spec': { fields: [], 'spec-id': 0, }, diff --git a/docker-compose.yml b/docker-compose.yml index a9e04eb73..b7fb27f8d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -47,10 +47,12 @@ services: S3_PROTOCOL_ACCESS_KEY_SECRET: 850181e4652dd023b7a98c58ae0d2d34bd487ee0cc3254aed6eda37307425907 # Iceberg Protocol ICEBERG_BUCKET_DETECTION_MODE: "FULL_PATH" - ICEBERG_WAREHOUSE: . - ICEBERG_CATALOG_URL: http://rest-catalog:8181/v1 - ICEBERG_CATALOG_AUTH_TYPE: token - ICEBERG_CATALOG_AUTH_TOKEN: token + ICEBERG_WAREHOUSE: warehouse--table-s3 + ICEBERG_SHARDS: warehouse--table-s3,warehouse-shard-2--table-s3 + ICEBERG_CATALOG_URL: http://rustfs:9000/iceberg/v1 + ICEBERG_CATALOG_AUTH_TYPE: sigv4 + ICEBERG_CATALOG_SIGV4_SERVICE: s3 + ICEBERG_CATALOG_PURGE_ON_DROP: false ICEBERG_S3_DELETE_ENABLED: true tenant_db: @@ -81,13 +83,6 @@ services: service: imgproxy file: ./.docker/docker-compose-infra.yml - rest-catalog: - depends_on: - rustfs_setup: - condition: service_completed_successfully - extends: - service: rest-catalog - file: ./.docker/docker-compose-infra.yml # Optional for rate-limiting # redis: # extends: diff --git a/src/config.ts b/src/config.ts index 3e72af1d3..6e36f94fd 100644 --- a/src/config.ts +++ b/src/config.ts @@ -270,6 +270,8 @@ type StorageConfigType = { icebergShards: string[] icebergCatalogUrl: string icebergCatalogAuthType: IcebergCatalogAuthType + icebergCatalogSigV4Service: string + icebergCatalogPurgeOnDrop: boolean icebergCatalogToken?: string icebergMaxNamespaceCount: number icebergMaxTableCount: number @@ -901,6 +903,10 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { icebergBucketDetectionMode: getOptionalConfigFromEnv('ICEBERG_BUCKET_DETECTION_MODE') || `BUCKET`, icebergCatalogAuthType: getOptionalConfigFromEnv('ICEBERG_CATALOG_AUTH_TYPE') || `sigv4`, + icebergCatalogSigV4Service: + getOptionalConfigFromEnv('ICEBERG_CATALOG_SIGV4_SERVICE') || 's3tables', + icebergCatalogPurgeOnDrop: + getOptionalConfigFromEnv('ICEBERG_CATALOG_PURGE_ON_DROP') !== 'false', icebergCatalogToken: getOptionalConfigFromEnv('ICEBERG_CATALOG_AUTH_TOKEN'), icebergMaxCatalogsCount: parseInt(getOptionalConfigFromEnv('ICEBERG_MAX_CATALOGS') || '2', 10), icebergMaxNamespaceCount: parseInt( diff --git a/src/storage/protocols/iceberg/catalog/index.ts b/src/storage/protocols/iceberg/catalog/index.ts index a30341173..69780516c 100644 --- a/src/storage/protocols/iceberg/catalog/index.ts +++ b/src/storage/protocols/iceberg/catalog/index.ts @@ -6,11 +6,11 @@ import { import { getConfig } from '../../../../config' export function getCatalogAuthStrategy(authType: string): CatalogAuthType { - const { storageS3Region, icebergCatalogToken } = getConfig() + const { storageS3Region, icebergCatalogToken, icebergCatalogSigV4Service } = getConfig() switch (authType) { case 'sigv4': - return new SignV4Auth({ region: storageS3Region }) + return new SignV4Auth({ region: storageS3Region, service: icebergCatalogSigV4Service }) case 'token': if (!icebergCatalogToken) { throw new Error('Iceberg catalog token is not configured') diff --git a/src/storage/protocols/iceberg/catalog/rest-catalog-client.test.ts b/src/storage/protocols/iceberg/catalog/rest-catalog-client.test.ts index d0befb6a1..ebca487e3 100644 --- a/src/storage/protocols/iceberg/catalog/rest-catalog-client.test.ts +++ b/src/storage/protocols/iceberg/catalog/rest-catalog-client.test.ts @@ -2,6 +2,7 @@ import { ErrorCode, type StorageBackendError } from '@internal/errors' import type { SignRequestOptions } from 'aws-sigv4-sign' import JSONBigint from 'json-bigint' import { beforeEach, describe, expect, it, vi } from 'vitest' +import { getConfig } from '../../../../config' import { spyOnAbortSignalTimeout } from '../../../../test/utils/abort-signal' const mockSignRequest = vi.fn() @@ -105,6 +106,60 @@ describe('RestCatalogClient request pipeline', () => { expect(fetchedUrl).toBe(signedUrl) }) + it('signs for the s3tables service by default and for a configured service', async () => { + const url = 'https://s3tables.ap-southeast-1.amazonaws.com/iceberg/v1' + + await new RestCatalogClient({ + catalogUrl: url, + auth: new SignV4Auth({ region: 'us-east-1' }), + }).getConfig({ warehouse: 'wh' }) + await new RestCatalogClient({ + catalogUrl: url, + auth: new SignV4Auth({ region: 'us-east-1', service: 's3' }), + }).getConfig({ warehouse: 'wh' }) + + expect(mockSignRequest.mock.calls[0][2]).toMatchObject({ service: 's3tables' }) + expect(mockSignRequest.mock.calls[1][2]).toMatchObject({ service: 's3' }) + }) + + it('sends purgeRequested on dropTable only when asked to purge and purging is enabled', async () => { + const client = new RestCatalogClient({ + catalogUrl: 'https://catalog.example.com/iceberg/v1', + auth: new BearerTokenAuth({ token: 'token' }), + }) + const dropUrl = () => new URL(fetchMock.mock.calls.at(-1)?.[0] as string) + + try { + vi.stubEnv('ICEBERG_CATALOG_PURGE_ON_DROP', 'true') + getConfig({ reload: true }) + + await client.dropTable({ warehouse: 'wh', namespace: 'ns', table: 'tbl' }) + expect(dropUrl().searchParams.has('purgeRequested')).toBe(false) + + await client.dropTable({ + warehouse: 'wh', + namespace: 'ns', + table: 'tbl', + purgeRequested: true, + }) + expect(dropUrl().searchParams.get('purgeRequested')).toBe('true') + + // Catalogs such as RustFS reject purgeRequested=true + vi.stubEnv('ICEBERG_CATALOG_PURGE_ON_DROP', 'false') + getConfig({ reload: true }) + await client.dropTable({ + warehouse: 'wh', + namespace: 'ns', + table: 'tbl', + purgeRequested: true, + }) + expect(dropUrl().searchParams.has('purgeRequested')).toBe(false) + } finally { + vi.unstubAllEnvs() + getConfig({ reload: true }) + } + }) + it('produces identical signed and fetched params even when params include empty strings', async () => { const client = new RestCatalogClient({ catalogUrl: 'https://host.example.com/v1', diff --git a/src/storage/protocols/iceberg/catalog/rest-catalog-client.ts b/src/storage/protocols/iceberg/catalog/rest-catalog-client.ts index 3cbcdcf8d..fbe2f5cca 100644 --- a/src/storage/protocols/iceberg/catalog/rest-catalog-client.ts +++ b/src/storage/protocols/iceberg/catalog/rest-catalog-client.ts @@ -1,6 +1,7 @@ import { ERRORS, StorageBackendError } from '@internal/errors' import { type SignRequestOptions, signRequest } from 'aws-sigv4-sign' import JSONBigint from 'json-bigint' +import { getConfig } from '../../../../config' import { createAlreadyExistsError, createAuthenticationTimeoutError, @@ -911,7 +912,9 @@ export class RestCatalogClient { const warehouse = this.getEncodedWarehouse(params.warehouse) const query: Record = {} - if (params.purgeRequested) { + // AWS S3 Tables only accepts purged drops, while catalogs such as RustFS reject + // `purgeRequested=true`. ICEBERG_CATALOG_PURGE_ON_DROP=false drops the parameter for the latter. + if (params.purgeRequested && getConfig().icebergCatalogPurgeOnDrop) { query.purgeRequested = 'true' } @@ -972,11 +975,14 @@ export class RestCatalogClient { * SignV4Auth class for AWS Signature Version 4 authentication * This class implements the CatalogAuth interface * to sign requests to the S3Tables service. + * S3-compatible catalogs that are not AWS S3 Tables (e.g. RustFS) can sign with a different + * service name via `service`. */ export class SignV4Auth { constructor( private readonly opts: { region: string + service?: string credentials?: SignRequestOptions['credentials'] } ) {} @@ -990,7 +996,7 @@ export class SignV4Auth { body: req.body, }, { - service: 's3tables', + service: this.opts.service ?? 's3tables', region: this.opts.region, credentials: this.opts.credentials, }