fix: Make sys.server_properties table filterable and resilient to unreachable servers

docs/querying/sql-metadata-tables.md

@@ -322,7 +322,7 @@ SELECT * FROM sys.supervisors WHERE healthy=0;
 
 ### SERVER_PROPERTIES table
 
-The `server_properties` table exposes the runtime properties configured on for each Druid server. Each row represents a single property key-value pair associated with a specific server.
+The `server_properties` table exposes the runtime properties configured on each Druid server. Each row represents a single property key-value pair associated with a specific server. This table supports filter and projection pushdown for efficient querying. If a server is unreachable, the table still returns a row for that server with the `error_message` column populated instead of failing the entire query.
 
 |Column|Type|Notes|
 |------|-----|-----|
@@ -331,6 +331,7 @@ The `server_properties` table exposes the runtime properties configured on for e
 |node_roles|VARCHAR|Comma-separated list of roles that the server performs. For example, `[coordinator,overlord]` if the server functions as both a Coordinator and an Overlord.|
 |property|VARCHAR|Name of the property|
 |value|VARCHAR|Value of the property|
+|error_message|VARCHAR|Describes why properties could not be retrieved from the server (e.g., connection refused, HTTP error). Null when properties were fetched successfully.|
 
 For example, to retrieve properties for a specific server, use the query
 

embedded-tests/src/test/java/org/apache/druid/testing/embedded/schema/SystemServerPropertiesTableTest.java

@@ -119,14 +119,14 @@ public void test_serverPropertiesTable_specificProperty()
     );
 
     Assertions.assertEquals(
-        StringUtils.format("localhost:%s,%s,[%s],test.onlyBroker,brokerValue", BROKER_PORT, BROKER_SERVICE, NodeRole.BROKER_JSON_NAME),
+        StringUtils.format("localhost:%s,%s,[%s],test.onlyBroker,brokerValue,", BROKER_PORT, BROKER_SERVICE, NodeRole.BROKER_JSON_NAME),
         cluster.runSql("SELECT * FROM sys.server_properties WHERE server = 'localhost:%s' AND property = 'test.onlyBroker'", BROKER_PORT)
     );
 
     String[] expectedRows = new String[] {
-        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,brokerNonUniqueValue", BROKER_PORT, BROKER_SERVICE, NodeRole.BROKER_JSON_NAME),
-        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,overlordNonUniqueValue", OVERLORD_PORT, OVERLORD_SERVICE, NodeRole.OVERLORD_JSON_NAME),
-        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,coordinatorNonUniqueValue", COORDINATOR_PORT, COORDINATOR_SERVICE, NodeRole.COORDINATOR_JSON_NAME),
+        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,brokerNonUniqueValue,", BROKER_PORT, BROKER_SERVICE, NodeRole.BROKER_JSON_NAME),
+        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,overlordNonUniqueValue,", OVERLORD_PORT, OVERLORD_SERVICE, NodeRole.OVERLORD_JSON_NAME),
+        StringUtils.format("localhost:%s,%s,[%s],test.nonUniqueProperty,coordinatorNonUniqueValue,", COORDINATOR_PORT, COORDINATOR_SERVICE, NodeRole.COORDINATOR_JSON_NAME),
     };
     Arrays.sort(expectedRows, String::compareTo);
     final String result = cluster.runSql("SELECT * FROM sys.server_properties WHERE property='test.nonUniqueProperty'");
@@ -146,6 +146,118 @@ public void test_serverPropertiesTable_hiddenProperties()
     Assertions.assertFalse(brokerProps.containsKey("password"));
   }
 
+  @Test
+  public void test_serverPropertiesTable_serverFilterPushdown()
+  {
+    final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT);
+
+    // Equality filter returns only matching server rows
+    final String result = cluster.runSql(
+        "SELECT server, service_name, property FROM sys.server_properties WHERE server = '%s'",
+        brokerHost
+    );
+    Assertions.assertFalse(result.isEmpty(), "Should return properties for the broker");
+    for (String row : result.split("\n")) {
+      Assertions.assertTrue(
+          row.startsWith(brokerHost + ","),
+          "Row should belong to filtered server: " + row
+      );
+    }
+
+    // Non-existent server returns no rows
+    final String emptyResult = cluster.runSql(
+        "SELECT * FROM sys.server_properties WHERE server = 'nonexistent:9999'"
+    );
+    Assertions.assertTrue(emptyResult.isEmpty(), "Non-existent server filter should return no rows");
+
+    // != is not consumed — falls back to Calcite post-filter, still correct
+    final String neResult = cluster.runSql(
+        "SELECT DISTINCT server FROM sys.server_properties WHERE server != '%s'",
+        brokerHost
+    );
+    Assertions.assertFalse(neResult.isEmpty(), "!= filter should still return other servers");
+    for (String row : neResult.split("\n")) {
+      Assertions.assertFalse(
+          row.trim().equals(brokerHost),
+          "!= filter should exclude the broker: " + row
+      );
+    }
+
+    // AND with a non-pushdown predicate — server filter consumed, rest handled by Calcite
+    final String andResult = cluster.runSql(
+        "SELECT server, property FROM sys.server_properties WHERE server = '%s' AND node_roles LIKE '%%broker%%'",
+        brokerHost
+    );
+    Assertions.assertFalse(andResult.isEmpty(), "AND with node_roles filter should return rows");
+    for (String row : andResult.split("\n")) {
+      Assertions.assertTrue(
+          row.startsWith(brokerHost + ","),
+          "Row should belong to filtered server: " + row
+      );
+    }
+  }
+
+  @Test
+  public void test_serverPropertiesTable_serviceNameFilterPushdown()
+  {
+    final String brokerHost = StringUtils.format("localhost:%s", BROKER_PORT);
+
+    // Equality filter on service_name returns only matching rows
+    final String result = cluster.runSql(
+        "SELECT server, service_name, property FROM sys.server_properties WHERE service_name = '%s'",
+        BROKER_SERVICE
+    );
+    Assertions.assertFalse(result.isEmpty(), "Should return properties for the broker service");
+    for (String row : result.split("\n")) {
+      String[] cols = row.split(",", -1);
+      Assertions.assertEquals(BROKER_SERVICE, cols[1], "Row should belong to filtered service_name: " + row);
+    }
+
+    // Non-existent service_name returns no rows
+    final String emptyResult = cluster.runSql(
+        "SELECT * FROM sys.server_properties WHERE service_name = 'nonexistent/service'"
+    );
+    Assertions.assertTrue(emptyResult.isEmpty(), "Non-existent service_name filter should return no rows");
+
+    // != falls back to Calcite post-filter
+    final String neResult = cluster.runSql(
+        "SELECT DISTINCT service_name FROM sys.server_properties WHERE service_name != '%s'",
+        BROKER_SERVICE
+    );
+    Assertions.assertFalse(neResult.isEmpty(), "!= filter should still return other services");
+    for (String row : neResult.split("\n")) {
+      Assertions.assertFalse(
+          row.trim().equals(BROKER_SERVICE),
+          "!= filter should exclude the broker service: " + row
+      );
+    }
+
+    // Both server and service_name filters consumed together
+    final String andResult = cluster.runSql(
+        "SELECT server, service_name, property FROM sys.server_properties WHERE service_name = '%s' AND server = '%s'",
+        BROKER_SERVICE, brokerHost
+    );
+    Assertions.assertFalse(andResult.isEmpty(), "AND with server and service_name should return rows");
+    for (String row : andResult.split("\n")) {
+      String[] cols = row.split(",", -1);
+      Assertions.assertEquals(brokerHost, cols[0], "Row server should match: " + row);
+      Assertions.assertEquals(BROKER_SERVICE, cols[1], "Row service_name should match: " + row);
+    }
+  }
+
+  @Test
+  public void test_serverPropertiesTable_errorMessageIsNullForHealthyServers()
+  {
+    // All 3 servers in the embedded cluster are healthy, so no rows should have a non-null error_message
+    final String errorRows = cluster.runSql("SELECT server FROM sys.server_properties WHERE error_message IS NOT NULL");
+    Assertions.assertTrue(errorRows.isEmpty(), "Healthy servers should have null error_message");
+
+    // Every row should have a null error_message
+    final String totalCount = cluster.runSql("SELECT COUNT(*) FROM sys.server_properties");
+    final String nullErrorCount = cluster.runSql("SELECT COUNT(*) FROM sys.server_properties WHERE error_message IS NULL");
+    Assertions.assertEquals(totalCount, nullErrorCount, "All rows should have null error_message in a healthy cluster");
+  }
+
   private void verifyPropertiesForServer(Map<String, String> properties, String serivceName, String hostAndPort, String nodeRole)
   {
     String[] expectedRows = properties.entrySet().stream().map(entry -> String.join(
@@ -154,7 +266,8 @@ private void verifyPropertiesForServer(Map<String, String> properties, String se
         escapeCsvField(serivceName),
         escapeCsvField(ImmutableList.of(nodeRole).toString()),
         escapeCsvField(entry.getKey()),
-        escapeCsvField(entry.getValue())
+        escapeCsvField(entry.getValue()),
+        escapeCsvField(null)
     )).toArray(String[]::new);
     Arrays.sort(expectedRows, String::compareTo);
     final String result = cluster.runSql("SELECT * FROM sys.server_properties WHERE server='%s'", hostAndPort);