Add validation for * column for non-COUNT function in star tree config (#17008)

* Add validation for * column for non-COUNT function in star tree config

* Add validation for aggregationConfig in startree config and related tests

Author: Sonam Mandal

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java

@@ -1331,6 +1331,9 @@ private static void validateStarTreeIndexConfigs(List<StarTreeIndexConfig> starT
           String column = columnPair.getColumn();
           if (!column.equals(AggregationFunctionColumnPair.STAR)) {
             referencedColumns.add(column);
+          } else if (columnPair.getFunctionType() != AggregationFunctionType.COUNT) {
+            throw new IllegalStateException("Non-COUNT function set the column as '*' in the functionColumnPair: "
+                + functionColumnPair + ". Please configure an actual column for the function");
           }
         }
       }
@@ -1357,6 +1360,10 @@ private static void validateStarTreeIndexConfigs(List<StarTreeIndexConfig> starT
           String column = columnPair.getColumn();
           if (!column.equals(AggregationFunctionColumnPair.STAR)) {
             referencedColumns.add(column);
+          } else if (columnPair.getFunctionType() != AggregationFunctionType.COUNT) {
+            throw new IllegalStateException("Non-COUNT function set the column as '*' in the aggregationConfig for "
+                + "function: " + aggregationConfig.getAggregationFunction()
+                + ". Please configure an actual column for the function");
           }
         }
       }

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java

@@ -1632,6 +1632,31 @@ public void testValidateIndexingConfig() {
       fail("Should not fail for valid StarTreeIndex config column name");
     }
 
+    // Test using * as the column for the COUNT aggregation
+    starTreeIndexConfig =
+        new StarTreeIndexConfig(List.of("myCol"), List.of("myCol"), List.of("COUNT__*"), null, 1);
+    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+        .build();
+    try {
+      TableConfigUtils.validate(tableConfig, schema);
+    } catch (Exception e) {
+      fail("Should not fail for valid StarTreeIndex function column pair with COUNT__*");
+    }
+
+    // Test using * as the column for a non-COUNT aggregation
+    starTreeIndexConfig =
+        new StarTreeIndexConfig(List.of("myCol"), List.of("myCol"), List.of("SUM__*"), null, 1);
+    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+        .build();
+    try {
+      TableConfigUtils.validate(tableConfig, schema);
+      fail("Should not fail for invalid StarTreeIndex function column pair with SUM__*");
+    } catch (Exception e) {
+      // expected
+    }
+
     starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol2"), List.of("myCol"), List.of("SUM__myCol"), null, 1);
     tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
         .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
@@ -1690,6 +1715,29 @@ public void testValidateIndexingConfig() {
       // expected
     }
 
+    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), null, null,
+        List.of(new StarTreeAggregationConfig("*", "COUNT")), 1);
+    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+        .build();
+    try {
+      TableConfigUtils.validate(tableConfig, schema);
+    } catch (Exception e) {
+      fail("Should not fail for valid StarTreeIndex config with aggregation config for COUNT on '*' column");
+    }
+
+    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), null, null,
+        List.of(new StarTreeAggregationConfig("*", "SUM")), 1);
+    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+        .build();
+    try {
+      TableConfigUtils.validate(tableConfig, schema);
+      fail("Should fail for invalid StarTreeIndex config with aggregation config for SUM on '*' column");
+    } catch (Exception e) {
+      // expected
+    }
+
     starTreeIndexConfig =
         new StarTreeIndexConfig(List.of("multiValCol"), List.of("multiValCol"), List.of("SUM__multiValCol"), null, 1);
     tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)