chore: use rustfs for iceberg rest catalog

.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

.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"
 

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,
     },

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:

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(

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')

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',

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<string, string> = {}
 
-    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,
       }