feat: Add configurable S3 metadata headers to catalog

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/aws/IceAwsClientFactory.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.altinity.ice.rest.catalog.internal.aws;
+
+import java.util.Map;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.HttpClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOAwsClientFactory;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import software.amazon.awssdk.core.client.builder.SdkClientBuilder;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
+
+/**
+ * S3FileIOAwsClientFactory to be set as {@link S3FileIOProperties#CLIENT_FACTORY}
+ * (s3.client-factory-impl) so that ice controls how S3 clients are built.
+ */
+public class IceAwsClientFactory implements S3FileIOAwsClientFactory {
+
+  private AwsClientProperties awsClientProperties;
+  private S3FileIOProperties s3FileIOProperties;
+  private HttpClientProperties httpClientProperties;
+  private Map<String, String> objectMetadata;
+
+  public IceAwsClientFactory() {
+    this.awsClientProperties = new AwsClientProperties();
+    this.s3FileIOProperties = new S3FileIOProperties();
+    this.httpClientProperties = new HttpClientProperties();
+    this.objectMetadata = Map.of();
+  }
+
+  @Override
+  public S3Client s3() {
+    return S3Client.builder()
+        .applyMutation(awsClientProperties::applyClientRegionConfiguration)
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
+        .applyMutation(s3FileIOProperties::applyEndpointConfigurations)
+        .applyMutation(s3FileIOProperties::applyServiceConfigurations)
+        .applyMutation(
+            b -> s3FileIOProperties.applyCredentialConfigurations(awsClientProperties, b))
+        .applyMutation(s3FileIOProperties::applySignerConfiguration)
+        .applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
+        .applyMutation(s3FileIOProperties::applyUserAgentConfigurations)
+        .applyMutation(s3FileIOProperties::applyRetryConfigurations)
+        .applyMutation(this::applyObjectMetadataConfiguration)
+        .build();
+  }
+
+  @Override
+  public S3AsyncClient s3Async() {
+    if (s3FileIOProperties.isS3CRTEnabled()) {
+      return S3AsyncClient.crtBuilder()
+          .applyMutation(awsClientProperties::applyClientRegionConfiguration)
+          .applyMutation(awsClientProperties::applyClientCredentialConfigurations)
+          .applyMutation(s3FileIOProperties::applyEndpointConfigurations)
+          .applyMutation(s3FileIOProperties::applyS3CrtConfigurations)
+          .build();
+    }
+    return S3AsyncClient.builder()
+        .applyMutation(awsClientProperties::applyClientRegionConfiguration)
+        .applyMutation(awsClientProperties::applyClientCredentialConfigurations)
+        .applyMutation(s3FileIOProperties::applyEndpointConfigurations)
+        .applyMutation(this::applyObjectMetadataConfiguration)
+        .build();
+  }
+
+  private void applyObjectMetadataConfiguration(SdkClientBuilder<?, ?> builder) {
+    if (objectMetadata.isEmpty()) {
+      return;
+    }
+    builder.overrideConfiguration(
+        c -> c.addExecutionInterceptor(new S3ObjectMetadataInterceptor(objectMetadata)));
+  }
+
+  @Override
+  public void initialize(Map<String, String> properties) {
+    this.awsClientProperties = new AwsClientProperties(properties);
+    this.s3FileIOProperties = new S3FileIOProperties(properties);
+    this.httpClientProperties = new HttpClientProperties(properties);
+    this.objectMetadata = S3ObjectMetadataInterceptor.metadataFromProperties(properties);
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/aws/S3ObjectMetadataInterceptor.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.altinity.ice.rest.catalog.internal.aws;
+
+import com.altinity.ice.internal.strings.Strings;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.Context;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+/**
+ * Adds user-defined object metadata (sent as x-amz-meta-* headers) to every S3 request that creates
+ * an object.
+ */
+public final class S3ObjectMetadataInterceptor implements ExecutionInterceptor {
+
+  public static final String METADATA_PREFIX = "s3.metadata.";
+
+  private static final String HEADER_PREFIX = "x-amz-meta-";
+
+  private final Map<String, String> metadata;
+
+  public S3ObjectMetadataInterceptor(Map<String, String> metadata) {
+    this.metadata = Map.copyOf(metadata);
+  }
+
+  /**
+   * Extracts object metadata from catalog properties. Keys are stripped of METADATA_PREFIX (both
+   * x-amz-meta-foo and foo result in the x-amz-meta-foo header).
+   */
+  public static Map<String, String> metadataFromProperties(Map<String, String> properties) {
+    Map<String, String> m = new LinkedHashMap<>();
+    for (Map.Entry<String, String> e : properties.entrySet()) {
+      String k = e.getKey();
+      if (!k.startsWith(METADATA_PREFIX)) {
+        continue;
+      }
+      k = Strings.removePrefix(k, METADATA_PREFIX);
+      if (k.toLowerCase().startsWith(HEADER_PREFIX)) {
+        k = k.substring(HEADER_PREFIX.length());
+      }
+      if (k.isEmpty() || e.getValue() == null) {
+        continue;
+      }
+      m.put(k, e.getValue());
+    }
+    return m;
+  }
+
+  @Override
+  public SdkRequest modifyRequest(Context.ModifyRequest context, ExecutionAttributes attrs) {
+    SdkRequest request = context.request();
+    if (request instanceof PutObjectRequest r) {
+      return r.toBuilder().metadata(merge(r.hasMetadata() ? r.metadata() : Map.of())).build();
+    }
+    if (request instanceof CreateMultipartUploadRequest r) {
+      return r.toBuilder().metadata(merge(r.hasMetadata() ? r.metadata() : Map.of())).build();
+    }
+    return request;
+  }
+
+  private Map<String, String> merge(Map<String, String> requestMetadata) {
+    Map<String, String> m = new HashMap<>(metadata);
+    m.putAll(requestMetadata);
+    return m;
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/Config.java

@@ -13,6 +13,8 @@
 import com.altinity.ice.internal.iceberg.io.SchemeFileIO;
 import com.altinity.ice.internal.strings.Strings;
 import com.altinity.ice.rest.catalog.internal.aws.CustomS3TablesCatalog;
+import com.altinity.ice.rest.catalog.internal.aws.IceAwsClientFactory;
+import com.altinity.ice.rest.catalog.internal.aws.S3ObjectMetadataInterceptor;
 import com.altinity.ice.rest.catalog.internal.etcd.EtcdCatalog;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -127,7 +129,26 @@ public record S3(
           String secretAccessKey,
       @JsonPropertyDescription(
               "AWS_REGION (see https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-envvars.html#envvars-list)")
-          String region) {}
+          String region,
+      @JsonPropertyDescription(
+              "User-defined metadata to attach to every object created by the catalog, e.g. \"x-amz-meta-expiration-seconds: 1000\" (the x-amz-meta- prefix is optional). Empty by default")
+          Map<String, String> metadata) {
+
+    public S3(
+        String endpoint,
+        boolean pathStyleAccess,
+        String accessKeyID,
+        String secretAccessKey,
+        String region,
+        Map<String, String> metadata) {
+      this.endpoint = endpoint;
+      this.pathStyleAccess = pathStyleAccess;
+      this.accessKeyID = accessKeyID;
+      this.secretAccessKey = secretAccessKey;
+      this.region = region;
+      this.metadata = Objects.requireNonNullElse(metadata, Map.of());
+    }
+  }
 
   public record Token(
       @JsonPropertyDescription("Name") String name,
@@ -249,6 +270,13 @@ public void putNotNullOrEmpty(String key, String value) {
       if (s3.pathStyleAccess) {
         m.putNotNullOrEmpty(S3FileIOProperties.PATH_STYLE_ACCESS, "true");
       }
+      if (!s3.metadata.isEmpty()) {
+        for (Map.Entry<String, String> e : s3.metadata.entrySet()) {
+          m.putNotNullOrEmpty(
+              S3ObjectMetadataInterceptor.METADATA_PREFIX + e.getKey(), e.getValue());
+        }
+        m.put(S3FileIOProperties.CLIENT_FACTORY, IceAwsClientFactory.class.getName());
+      }
     }
 
     if (localFileIOBaseDir != null) {

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/RESTCatalogTestBase.java

@@ -95,7 +95,7 @@ public void setUp() throws Exception {
             "jdbc:sqlite::memory:", // uri
             "s3://test-bucket/warehouse", // warehouse
             null, // localFileIOBaseDir
-            new Config.S3(minioEndpoint, true, "minioadmin", "minioadmin", "us-east-1"), // s3
+            new Config.S3(minioEndpoint, true, "minioadmin", "minioadmin", "us-east-1", null), // s3
             null, // bearerTokens
             new Config.AnonymousAccess(
                 true,

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/internal/aws/S3ObjectMetadataInterceptorTest.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.altinity.ice.rest.catalog.internal.aws;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.junit.Test;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+public class S3ObjectMetadataInterceptorTest {
+
+  @Test
+  public void metadataFromPropertiesIgnoresUnrelatedPropertiesAndHeaderPrefix() {
+    var m =
+        S3ObjectMetadataInterceptor.metadataFromProperties(
+            Map.of(
+                "s3.endpoint", "http://localhost:9000",
+                "s3.metadata.x-amz-meta-expiration-seconds", "1000",
+                "s3.metadata.owner", "ice",
+                "s3.metadata.", "ignored"));
+    assertThat(m).containsOnly(Map.entry("expiration-seconds", "1000"), Map.entry("owner", "ice"));
+  }
+
+  @Test
+  public void putObjectAndCreateMultipartUploadGetMetadata() {
+    var i = new S3ObjectMetadataInterceptor(Map.of("owner", "ice"));
+
+    var put = (PutObjectRequest) modify(i, PutObjectRequest.builder().bucket("b").key("k").build());
+    assertThat(put.metadata()).containsExactly(Map.entry("owner", "ice"));
+
+    var mpu =
+        (CreateMultipartUploadRequest)
+            modify(i, CreateMultipartUploadRequest.builder().bucket("b").key("k").build());
+    assertThat(mpu.metadata()).containsExactly(Map.entry("owner", "ice"));
+  }
+
+  @Test
+  public void requestMetadataWins() {
+    var i = new S3ObjectMetadataInterceptor(Map.of("owner", "ice", "env", "prod"));
+    var put =
+        (PutObjectRequest)
+            modify(
+                i,
+                PutObjectRequest.builder()
+                    .bucket("b")
+                    .key("k")
+                    .metadata(Map.of("owner", "explicit"))
+                    .build());
+    assertThat(put.metadata())
+        .containsOnly(Map.entry("owner", "explicit"), Map.entry("env", "prod"));
+  }
+
+  @Test
+  public void otherRequestsAreLeftAlone() {
+    var i = new S3ObjectMetadataInterceptor(Map.of("owner", "ice"));
+    var get = GetObjectRequest.builder().bucket("b").key("k").build();
+    assertThat(modify(i, get)).isSameAs(get);
+  }
+
+  private static SdkRequest modify(S3ObjectMetadataInterceptor i, SdkRequest request) {
+    return i.modifyRequest(() -> request, new ExecutionAttributes());
+  }
+}