Fix JDBC hasChildren fetching all rows and columns

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -22,6 +22,7 @@
 
 import com.google.common.base.Preconditions;
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -68,6 +69,7 @@
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.StorageLocation;
+import org.apache.polaris.persistence.relational.jdbc.models.Converter;
 import org.apache.polaris.persistence.relational.jdbc.models.EntityNameLookupRecordConverter;
 import org.apache.polaris.persistence.relational.jdbc.models.EntityVersionConverter;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelCommitMetricsReport;
@@ -746,9 +748,19 @@ public boolean hasChildren(
     try {
       var results =
           datasourceOperations.executeSelect(
-              QueryGenerator.generateSelectQuery(
-                  ModelEntity.getAllColumnNames(schemaVersion), ModelEntity.TABLE_NAME, params, 1),
-              new ModelEntity(schemaVersion));
+              QueryGenerator.generateExistsQuery(
+                  ModelEntity.getAllColumnNames(schemaVersion), ModelEntity.TABLE_NAME, params),
+              new Converter<Integer>() {
+                @Override
+                public Integer fromResultSet(ResultSet rs) {
+                  return 1;
+                }
+
+                @Override
+                public Map<String, Object> toMap(DatabaseType databaseType) {
+                  throw new UnsupportedOperationException();
+                }
+              });
       return results != null && !results.isEmpty();
     } catch (SQLException e) {
       throw new RuntimeException(

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/QueryGenerator.java

@@ -413,6 +413,20 @@ static PreparedQuery generateVersionQuery() {
     return new PreparedQuery("SELECT version_value FROM POLARIS_SCHEMA.VERSION", List.of());
   }
 
+  /**
+   * Generates a {@code SELECT 1 ... WHERE ... LIMIT 1} query to test row existence without fetching
+   * any column data. All filter conditions must be supplied in {@code whereClause}.
+   */
+  public static PreparedQuery generateExistsQuery(
+      @NonNull List<String> tableColumns,
+      @NonNull String tableName,
+      @NonNull Map<String, Object> whereClause) {
+    QueryFragment where = generateWhereClause(new HashSet<>(tableColumns), whereClause, Map.of());
+    String sql =
+        "SELECT 1 FROM " + getFullyQualifiedTableName(tableName) + where.sql() + " LIMIT 1";
+    return new PreparedQuery(sql, where.parameters());
+  }
+
   @VisibleForTesting
   static PreparedQuery generateEntityTableExistQuery() {
     return new PreparedQuery(

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImplTest.java

@@ -217,6 +217,45 @@ void lookupEntityVersionsPreservesOrderAndNullsForMissing(int schemaVersion)
     assertThat(captor.getValue().sql()).doesNotContain("properties");
   }
 
+  @ParameterizedTest
+  @ValueSource(ints = {1, 2})
+  void hasChildrenUsesExistsQueryNotFullScan(int schemaVersion) throws SQLException, IOException {
+    JdbcConnectionPool dataSource =
+        JdbcConnectionPool.create(
+            "jdbc:h2:mem:has_children_v"
+                + schemaVersion
+                + "_"
+                + System.nanoTime()
+                + ";DB_CLOSE_DELAY=-1",
+            "sa",
+            "");
+    DatasourceOperations real = new DatasourceOperations(dataSource, new TestJdbcConfiguration());
+    try (InputStream script = DatabaseType.H2.openInitScriptResource(schemaVersion)) {
+      real.executeScript(script);
+    }
+    DatasourceOperations spy = Mockito.spy(real);
+    doCallRealMethod().when(spy).executeSelect(any(), any());
+
+    JdbcBasePersistenceImpl impl =
+        new JdbcBasePersistenceImpl(
+            new PolarisDefaultDiagServiceImpl(),
+            spy,
+            RANDOM_SECRETS,
+            REALM_CONTEXT.getRealmIdentifier(),
+            schemaVersion);
+    PolarisCallContext callCtx = new PolarisCallContext(REALM_CONTEXT, impl);
+
+    impl.hasChildren(callCtx, null, 0L, 0L);
+
+    ArgumentCaptor<QueryGenerator.PreparedQuery> captor =
+        ArgumentCaptor.forClass(QueryGenerator.PreparedQuery.class);
+    verify(spy).executeSelect(captor.capture(), any());
+    String sql = captor.getValue().sql();
+    assertThat(sql).contains("LIMIT 1");
+    assertThat(sql).doesNotContain("properties");
+    assertThat(sql).doesNotContain("internal_properties");
+  }
+
   private static final class TestJdbcConfiguration implements RelationalJdbcConfiguration {
     @Override
     public Optional<Integer> maxRetries() {

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/QueryGeneratorTest.java

@@ -19,7 +19,9 @@
 package org.apache.polaris.persistence.relational.jdbc;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -461,4 +463,33 @@ void testGenerateOverlapQuery() {
         .containsExactly(
             "realmId", -123L, "/", "//", "//バケツ/", "//バケツ/\"loc.ation\"/", "//バケツ/\"loc.ation\"/%");
   }
+
+  @Test
+  void testGenerateExistsQuery() {
+    Map<String, Object> params = new HashMap<>();
+    params.put("realm_id", "realm1");
+    params.put("catalog_id", 1L);
+    params.put("parent_id", 2L);
+    String sql =
+        QueryGenerator.generateExistsQuery(
+                ModelEntity.getAllColumnNames(2), ModelEntity.TABLE_NAME, params)
+            .sql();
+    assertTrue(sql.startsWith("SELECT 1 "), sql);
+    assertTrue(sql.endsWith("LIMIT 1"), sql);
+    assertFalse(sql.contains("properties"), sql);
+    assertTrue(sql.contains("realm_id"), sql);
+    assertTrue(sql.contains("catalog_id"), sql);
+    assertTrue(sql.contains("parent_id"), sql);
+  }
+
+  @Test
+  void testGenerateExistsQueryRejectsUnknownColumn() {
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            QueryGenerator.generateExistsQuery(
+                ModelEntity.getAllColumnNames(2),
+                ModelEntity.TABLE_NAME,
+                Map.of("not_a_column", 1)));
+  }
 }