fix(s3): strip endpoint scheme from env vars and honor ssl.enabled=true

ConfigureS3Options split the URI scheme off only for the s3.endpoint
property; the AWS_ENDPOINT_URL[_S3] env-var fallback used the raw value, so
a scheme-prefixed env endpoint (the AWS convention) reached Arrow's
endpoint_override unchanged. Factor the split into a helper applied to all
three sources.

Also make s3.ssl.enabled symmetric: an explicit true now forces the https
scheme (previously only false was honored), matching the comment.

Author: plusplusjiajia

src/iceberg/arrow/s3/arrow_s3_file_io.cc

@@ -75,6 +75,17 @@ Status EnsureS3Initialized() {
   return {};
 }
 
+// Splits any URI scheme off `endpoint` into `options.scheme`, returning the bare
+// host[:port] that Arrow's `endpoint_override` expects.
+std::string SplitEndpointScheme(std::string_view endpoint,
+                                ::arrow::fs::S3Options& options) {
+  if (const auto pos = endpoint.find("://"); pos != std::string_view::npos) {
+    options.scheme = std::string(endpoint.substr(0, pos));
+    endpoint = endpoint.substr(pos + 3);
+  }
+  return std::string(endpoint);
+}
+
 #endif
 
 }  // namespace
@@ -116,27 +127,17 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
     options.region = *region;
   }
 
-  // Configure endpoint (for MinIO, LocalStack, OSS, etc.)
+  // Configure endpoint (for MinIO, LocalStack, OSS, etc.) from `s3.endpoint` or
+  // the AWS standard env vars.
   if (const auto* endpoint = FindProperty(properties, S3Properties::kEndpoint);
       endpoint != nullptr) {
-    // `s3.endpoint` may be a full URI; Arrow wants host[:port], so strip the scheme.
-    std::string_view ep = *endpoint;
-    if (const auto pos = ep.find("://"); pos != std::string_view::npos) {
-      options.scheme = std::string(ep.substr(0, pos));
-      ep = ep.substr(pos + 3);
-    }
-    options.endpoint_override = std::string(ep);
-  } else {
-    // Fall back to AWS standard environment variables for endpoint override
-    const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");
-    if (s3_endpoint_env != nullptr) {
-      options.endpoint_override = s3_endpoint_env;
-    } else {
-      const char* endpoint_env = std::getenv("AWS_ENDPOINT_URL");
-      if (endpoint_env != nullptr) {
-        options.endpoint_override = endpoint_env;
-      }
-    }
+    options.endpoint_override = SplitEndpointScheme(*endpoint, options);
+  } else if (const char* s3_endpoint_env = std::getenv("AWS_ENDPOINT_URL_S3");
+             s3_endpoint_env != nullptr) {
+    options.endpoint_override = SplitEndpointScheme(s3_endpoint_env, options);
+  } else if (const char* endpoint_env = std::getenv("AWS_ENDPOINT_URL");
+             endpoint_env != nullptr) {
+    options.endpoint_override = SplitEndpointScheme(endpoint_env, options);
   }
 
   ICEBERG_ASSIGN_OR_RAISE(const auto path_style_access,
@@ -145,12 +146,11 @@ Result<::arrow::fs::S3Options> ConfigureS3Options(
     options.force_virtual_addressing = !*path_style_access;
   }
 
-  // Configure SSL. Explicit `s3.ssl.enabled` overrides any scheme derived from
-  // the endpoint above.
+  // Explicit `s3.ssl.enabled` overrides any endpoint-derived scheme.
   ICEBERG_ASSIGN_OR_RAISE(const auto ssl_enabled,
                           ParseOptionalBool(properties, S3Properties::kSslEnabled));
-  if (ssl_enabled.has_value() && !*ssl_enabled) {
-    options.scheme = "http";
+  if (ssl_enabled.has_value()) {
+    options.scheme = *ssl_enabled ? "https" : "http";
   }
 
   // Configure timeouts

src/iceberg/test/arrow_s3_file_io_test.cc

@@ -227,6 +227,20 @@ TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsKeepsSchemelessEndpoint) {
   EXPECT_EQ(result->endpoint_override, "localhost:9000");
 }
 
+TEST_F(ArrowS3FileIOTest, ConfigureS3OptionsSslEnabledOverridesEndpointScheme) {
+  auto https =
+      ConfigureS3Options({{std::string(S3Properties::kEndpoint), "http://localhost:9000"},
+                          {std::string(S3Properties::kSslEnabled), "true"}});
+  ASSERT_THAT(https, IsOk());
+  EXPECT_EQ(https->scheme, "https");
+
+  auto http = ConfigureS3Options(
+      {{std::string(S3Properties::kEndpoint), "https://localhost:9000"},
+       {std::string(S3Properties::kSslEnabled), "false"}});
+  ASSERT_THAT(http, IsOk());
+  EXPECT_EQ(http->scheme, "http");
+}
+
 TEST_F(ArrowS3FileIOTest,
        ConfigureS3OptionsPathStyleAccessFalseEnablesVirtualAddressing) {
   auto result =