Add HTTPS detection for ClickHouse JDBC URL parsing

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseJdbcUrlParser.java

@@ -117,6 +117,8 @@ static ParsedJdbcUrl parse(String jdbcUrl) {
   /**
    * Extracts and normalizes the HTTP/HTTPS URL from a JDBC URL.
    *
+   * <p>Automatically detects HTTPS based on port 8443 or ssl=true parameter.
+   *
    * @param jdbcUrl the JDBC URL to process
    * @return normalized HTTP/HTTPS URL
    * @throws IllegalArgumentException if the URL format is invalid
@@ -140,28 +142,41 @@ private static String extractHttpUrl(String jdbcUrl) {
               + jdbcUrl);
     }
 
-    // Check if URL already has a scheme and validate it
-    if (actualUrl.toLowerCase().startsWith("http://")
-        || actualUrl.toLowerCase().startsWith("https://")) {
-      return actualUrl;
+    boolean useHttps = false;
+
+    // Check if port suggests HTTPS (8443 is default HTTPS port for ClickHouse)
+    if (actualUrl.contains(":8443")) {
+      useHttps = true;
+    }
+
+    // Check if ssl=true in query string
+    if (actualUrl.toLowerCase().contains("ssl=true")) {
+      useHttps = true;
     }
 
     // Check for invalid schemes before prepending http://
     if (actualUrl.contains("://")) {
       // Extract the scheme part
       int schemeEnd = actualUrl.indexOf("://");
       String scheme = actualUrl.substring(0, schemeEnd).toLowerCase();
-      if (!scheme.equals("http") && !scheme.equals("https")) {
+
+      if (scheme.equals("http") || scheme.equals("https")) {
+        // If http:// but ssl=true detected, upgrade to https://
+        if (scheme.equals("http") && useHttps) {
+          actualUrl = "https://" + actualUrl.substring(schemeEnd + 3);
+        }
+        return actualUrl;
+      } else {
         throw new IllegalArgumentException(
             "Invalid scheme in JDBC URL. Expected 'http' or 'https'. Got: " + scheme);
       }
     }
 
-    // If URL doesn't start with http:// or https://, assume http://
+    // If URL doesn't start with http:// or https://, add the appropriate scheme
     if (actualUrl.startsWith("//")) {
-      actualUrl = "http:" + actualUrl;
+      actualUrl = (useHttps ? "https:" : "http:") + actualUrl;
     } else {
-      actualUrl = "http://" + actualUrl;
+      actualUrl = (useHttps ? "https://" : "http://") + actualUrl;
     }
 
     return actualUrl;

sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseJdbcUrlParserTest.java

@@ -338,4 +338,92 @@ public void testJdbcUrlWithEmptyQueryParameter() {
     assertEquals("", props.getProperty("user"));
     assertEquals("secret", props.getProperty("password"));
   }
+
+  @Test
+  public void testJdbcUrlWithSslParameter() {
+    String jdbcUrl = "jdbc:clickhouse://localhost:8443/mydb?ssl=true&user=admin";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("https://localhost:8443", parsed.getClickHouseUrl());
+    assertEquals("mydb", parsed.getDatabase());
+    assertEquals("admin", parsed.getProperties().getProperty("user"));
+    assertEquals("true", parsed.getProperties().getProperty("ssl"));
+  }
+
+  @Test
+  public void testJdbcUrlWithPort8443DefaultsToHttps() {
+    String jdbcUrl = "jdbc:clickhouse://myhost:8443/mydb";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("https://myhost:8443", parsed.getClickHouseUrl());
+    assertEquals("mydb", parsed.getDatabase());
+  }
+
+  @Test
+  public void testClickHouseCloudUrl() {
+    String jdbcUrl =
+        "jdbc:clickhouse://someservice.clickhouse.cloud:8443/default?"
+            + "user=default&password=secret&ssl=true&sslmode=NONE";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("https://someservice.clickhouse.cloud:8443", parsed.getClickHouseUrl());
+    assertEquals("default", parsed.getDatabase());
+    assertEquals("default", parsed.getProperties().getProperty("user"));
+    assertEquals("secret", parsed.getProperties().getProperty("password"));
+    assertEquals("true", parsed.getProperties().getProperty("ssl"));
+  }
+
+  @Test
+  public void testJdbcUrlPort8123DefaultsToHttp() {
+    String jdbcUrl = "jdbc:clickhouse://localhost:8123/mydb";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("http://localhost:8123", parsed.getClickHouseUrl());
+  }
+
+  @Test
+  public void testHttpSchemeUpgradedToHttpsWhenSslTrue() {
+    String jdbcUrl = "jdbc:clickhouse:http://localhost:8443/mydb?ssl=true";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    // Should upgrade http to https because ssl=true
+    assertEquals("https://localhost:8443", parsed.getClickHouseUrl());
+  }
+
+  @Test
+  public void testSslTrueTriggersHttps() {
+    String jdbcUrl = "jdbc:clickhouse://localhost:8123/mydb?ssl=true";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    // Should use https because ssl=true, even though port is 8123
+    assertEquals("https://localhost:8123", parsed.getClickHouseUrl());
+  }
+
+  @Test
+  public void testPort8443TriggersHttps() {
+    String jdbcUrl = "jdbc:clickhouse://localhost:8443/mydb";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    // Should use https because port is 8443
+    assertEquals("https://localhost:8443", parsed.getClickHouseUrl());
+  }
+
+  @Test
+  public void testClickHouseCloudUrlWithSsl() {
+    String jdbcUrl =
+        "jdbc:clickhouse://someservice.clickhouse.cloud:8443/default?"
+            + "user=default&password=secret&ssl=true&sslmode=NONE";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("https://someservice.clickhouse.cloud:8443", parsed.getClickHouseUrl());
+    assertEquals("default", parsed.getDatabase());
+  }
+
+  @Test
+  public void testExplicitHttpsPreserved() {
+    String jdbcUrl = "jdbc:clickhouse:https://localhost:8443/mydb";
+    ParsedJdbcUrl parsed = ClickHouseJdbcUrlParser.parse(jdbcUrl);
+
+    assertEquals("https://localhost:8443", parsed.getClickHouseUrl());
+  }
 }