Iceberg Bucket support

.docker/docker-compose-infra.yml

@@ -117,6 +117,10 @@ services:
     ports:
       - '9000:9000'
       - '9001:9001'
+    networks:
+      default:
+        aliases:
+          - warehouse--table-s3.minio
     healthcheck:
       test: timeout 5s bash -c ':> /dev/tcp/127.0.0.1/9000' || exit 1
       interval: 5s
@@ -125,6 +129,7 @@ services:
     environment:
       MINIO_ROOT_USER: supa-storage
       MINIO_ROOT_PASSWORD: secret1234
+      MINIO_DOMAIN: minio
     command: server --console-address ":9001" /data
 
   minio_setup:
@@ -136,6 +141,8 @@ services:
       /bin/sh -c "
       /usr/bin/mc alias set supa-minio http://minio:9000 supa-storage secret1234;
       /usr/bin/mc mb supa-minio/supa-storage-bucket;
+      /usr/bin/mc mb supa-minio/warehouse--table-s3;
+      /usr/bin/mc policy set public supa-minio/warehouse--table-s3;
       exit 0;
       "
 
@@ -153,6 +160,21 @@ services:
       - IMGPROXY_USE_ETAG=true
       - IMGPROXY_ENABLE_WEBP_DETECTION=true
 
+  rest-catalog:
+    image: tabulario/iceberg-rest
+    container_name: iceberg-rest
+    depends_on:
+      - minio_setup
+    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://minio:9000
+
 # Optional for rate-limiting
 #  redis:
 #    image: redis:6.2-alpine

migrations/multitenant/0019-iceberg-catalog-resources.sql

@@ -0,0 +1,39 @@
+ALTER TABLE tenants ADD COLUMN IF NOT EXISTS feature_iceberg_catalog boolean NOT NULL DEFAULT false;
+ALTER TABLE tenants ADD COLUMN IF NOT EXISTS feature_iceberg_catalog_max_namespaces int NOT NULL DEFAULT 10;
+ALTER TABLE tenants ADD COLUMN IF NOT EXISTS feature_iceberg_catalog_max_tables int NOT NULL DEFAULT 10;
+ALTER TABLE tenants ADD COLUMN IF NOT EXISTS feature_iceberg_catalog_max_catalogs int NOT NULL DEFAULT 2;
+
+CREATE TABLE IF NOT EXISTS iceberg_catalogs (
+  id text not null,
+  tenant_id text NOT NULL,
+  created_at timestamptz NOT NULL default now(),
+  updated_at timestamptz NOT NULL default now(),
+
+  primary key (id, tenant_id)
+);
+
+CREATE TABLE IF NOT EXISTS iceberg_namespaces (
+  id uuid primary key default gen_random_uuid(),
+  tenant_id text NOT NULL,
+  bucket_id text NOT NULL,
+  name text COLLATE "C" NOT NULL,
+  created_at timestamptz NOT NULL default now(),
+  updated_at timestamptz NOT NULL default now()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_iceberg_namespaces_bucket_id ON iceberg_namespaces (tenant_id, bucket_id, name);
+
+CREATE TABLE IF NOT EXISTS iceberg_tables (
+  id uuid primary key default gen_random_uuid(),
+  tenant_id text NOT NULL,
+  namespace_id uuid NOT NULL references iceberg_namespaces(id) ON DELETE CASCADE,
+  bucket_id text NOT NULL,
+  name text COLLATE "C" NOT NULL,
+  location text not null,
+  created_at timestamptz NOT NULL default now(),
+  updated_at timestamptz NOT NULL default now()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_iceberg_tables_tenant_namespace_id ON iceberg_tables (tenant_id, namespace_id, name);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_iceberg_tables_tenant_location ON iceberg_tables (tenant_id, location);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_iceberg_tables_location ON iceberg_tables (location);

migrations/tenant/0038-iceberg-catalog-flag-on-buckets.sql

@@ -0,0 +1,67 @@
+DO $$
+    DECLARE
+        is_multitenant bool = COALESCE(current_setting('storage.multitenant', true), 'false')::boolean;
+        anon_role text = COALESCE(current_setting('storage.anon_role', true), 'anon');
+        authenticated_role text = COALESCE(current_setting('storage.authenticated_role', true), 'authenticated');
+        service_role text = COALESCE(current_setting('storage.service_role', true), 'service_role');
+    BEGIN
+        IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'buckettype') THEN
+            create type storage.BucketType as enum (
+                'STANDARD',
+                'ANALYTICS'
+            );
+        END IF;
+
+        ALTER TABLE storage.buckets DROP COLUMN IF EXISTS iceberg_catalog;
+        ALTER TABLE storage.buckets ADD COLUMN IF NOT EXISTS type storage.BucketType NOT NULL default 'STANDARD';
+
+        CREATE TABLE IF NOT EXISTS storage.buckets_analytics (
+            id text not null primary key,
+            type storage.BucketType NOT NULL default 'ANALYTICS',
+            format text NOT NULL default 'ICEBERG',
+            created_at timestamptz NOT NULL default now(),
+            updated_at timestamptz NOT NULL default now()
+        );
+
+        ALTER TABLE storage.buckets_analytics ADD COLUMN IF NOT EXISTS type storage.BucketType NOT NULL default 'ANALYTICS';
+        ALTER TABLE storage.buckets_analytics ENABLE ROW LEVEL SECURITY;
+
+        EXECUTE 'GRANT ALL ON TABLE storage.buckets_analytics TO ' || service_role || ', ' || authenticated_role || ', ' || anon_role;
+
+        IF is_multitenant THEN
+            RETURN;
+        END IF;
+
+        CREATE TABLE IF NOT EXISTS storage.iceberg_namespaces (
+            id uuid primary key default gen_random_uuid(),
+            bucket_id text NOT NULL references storage.buckets_analytics(id) ON DELETE CASCADE,
+            name text COLLATE "C" NOT NULL,
+            created_at timestamptz NOT NULL default now(),
+            updated_at timestamptz NOT NULL default now()
+        );
+
+        CREATE UNIQUE INDEX IF NOT EXISTS idx_iceberg_namespaces_bucket_id ON storage.iceberg_namespaces (bucket_id, name);
+
+        CREATE TABLE IF NOT EXISTS storage.iceberg_tables (
+          id uuid primary key default gen_random_uuid(),
+          namespace_id uuid NOT NULL references storage.iceberg_namespaces(id) ON DELETE CASCADE,
+          bucket_id text NOT NULL references storage.buckets_analytics(id) ON DELETE CASCADE,
+          name text COLLATE "C" NOT NULL,
+          location text not null,
+          created_at timestamptz NOT NULL default now(),
+          updated_at timestamptz NOT NULL default now()
+        );
+
+        CREATE UNIQUE INDEX idx_iceberg_tables_namespace_id ON storage.iceberg_tables (namespace_id, name);
+
+        ALTER TABLE storage.iceberg_namespaces ENABLE ROW LEVEL SECURITY;
+        ALTER TABLE storage.iceberg_tables ENABLE ROW LEVEL SECURITY;
+
+        EXECUTE 'revoke all on storage.iceberg_namespaces from ' || anon_role || ', ' || authenticated_role;
+        EXECUTE 'GRANT ALL ON TABLE storage.iceberg_namespaces TO ' || service_role;
+        EXECUTE 'GRANT SELECT ON TABLE storage.iceberg_namespaces TO ' || authenticated_role || ', ' || anon_role;
+
+        EXECUTE 'revoke all on storage.iceberg_tables from ' || anon_role || ', ' || authenticated_role;
+        EXECUTE 'GRANT ALL ON TABLE storage.iceberg_tables TO ' || service_role;
+        EXECUTE 'GRANT SELECT ON TABLE storage.iceberg_tables TO ' || authenticated_role || ', ' || anon_role;
+END$$;