Handle null values appropriately during segment reload for new derived columns

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java

@@ -1485,6 +1485,7 @@ public void testStarTreeTriggering()
    *   <li>"NewAddedRawDerivedStringDimension", DIMENSION, STRING, single-value, reverse(DestCityName)</li>
    *   <li>"NewAddedRawDerivedMVIntDimension", DIMENSION, INT, multi-value, array(ActualElapsedTime)</li>
    *   <li>"NewAddedDerivedMVDoubleDimension", DIMENSION, DOUBLE, multi-value, array(ArrDelayMinutes)</li>
+   *   <li>"NewAddedDerivedNullString", DIMENSION, STRING, single-value, caseWhen(true, null, null)</li>
    * </ul>
    */
   @Test(dependsOnMethods = "testAggregateMetadataAPI", dataProvider = "useBothQueryEngines")
@@ -1496,7 +1497,7 @@ public void testDefaultColumns(boolean useMultiStageQueryEngine)
     reloadWithExtraColumns();
     JsonNode queryResponse = postQuery(SELECT_STAR_QUERY);
     assertEquals(queryResponse.get("totalDocs").asLong(), numTotalDocs);
-    assertEquals(queryResponse.get("resultTable").get("dataSchema").get("columnNames").size(), 103);
+    assertEquals(queryResponse.get("resultTable").get("dataSchema").get("columnNames").size(), 104);
 
     testNewAddedColumns();
 
@@ -1581,6 +1582,8 @@ private void reloadWithExtraColumns()
     schema.addField(new DimensionFieldSpec("NewAddedRawDerivedStringDimension", DataType.STRING, true));
     schema.addField(new DimensionFieldSpec("NewAddedRawDerivedMVIntDimension", DataType.INT, false));
     schema.addField(new DimensionFieldSpec("NewAddedDerivedMVDoubleDimension", DataType.DOUBLE, false));
+    schema.addField(new DimensionFieldSpec("NewAddedDerivedNullString", DataType.STRING, true, "nil"));
+    schema.setEnableColumnBasedNullHandling(true);
     addSchema(schema);
 
     TableConfig tableConfig = getOfflineTableConfig();
@@ -1593,7 +1596,8 @@ private void reloadWithExtraColumns()
             new TransformConfig("NewAddedDerivedDivAirportSeqIDsString", "DivAirportSeqIDs"),
             new TransformConfig("NewAddedRawDerivedStringDimension", "reverse(DestCityName)"),
             new TransformConfig("NewAddedRawDerivedMVIntDimension", "array(ActualElapsedTime)"),
-            new TransformConfig("NewAddedDerivedMVDoubleDimension", "array(ArrDelayMinutes)"));
+            new TransformConfig("NewAddedDerivedMVDoubleDimension", "array(ArrDelayMinutes)"),
+            new TransformConfig("NewAddedDerivedNullString", "caseWhen(true, null, null)"));
     IngestionConfig ingestionConfig = new IngestionConfig();
     ingestionConfig.setTransformConfigs(transformConfigs);
     tableConfig.setIngestionConfig(ingestionConfig);
@@ -1618,30 +1622,38 @@ private void reloadWithExtraColumns()
     JsonNode columnIndexSizeMap = JsonUtils.stringToJsonNode(sendGetRequest(
             getControllerBaseApiUrl() + "/tables/mytable/metadata?columns=DivAirportSeqIDs"
                 + "&columns=NewAddedDerivedDivAirportSeqIDs&columns=NewAddedDerivedDivAirportSeqIDsString"
-                + "&columns=NewAddedRawDerivedStringDimension&columns=NewAddedRawDerivedMVIntDimension"))
+                + "&columns=NewAddedRawDerivedStringDimension&columns=NewAddedRawDerivedMVIntDimension"
+                + "&columns=NewAddedDerivedNullString"))
         .get("columnIndexSizeMap");
-    assertEquals(columnIndexSizeMap.size(), 5);
+    assertEquals(columnIndexSizeMap.size(), 6);
     JsonNode originalColumnIndexSizes = columnIndexSizeMap.get("DivAirportSeqIDs");
     JsonNode derivedColumnIndexSizes = columnIndexSizeMap.get("NewAddedDerivedDivAirportSeqIDs");
     JsonNode derivedStringColumnIndexSizes = columnIndexSizeMap.get("NewAddedDerivedDivAirportSeqIDsString");
     JsonNode derivedRawStringColumnIndex = columnIndexSizeMap.get("NewAddedRawDerivedStringDimension");
     JsonNode derivedRawMVIntColumnIndex = columnIndexSizeMap.get("NewAddedRawDerivedMVIntDimension");
+    JsonNode derivedNullStringColumnIndex = columnIndexSizeMap.get("NewAddedDerivedNullString");
 
     // Derived int column should have the same dictionary size as the original column
-    double originalColumnDictionarySize = originalColumnIndexSizes.get("dictionary").asDouble();
-    assertEquals(derivedColumnIndexSizes.get("dictionary").asDouble(), originalColumnDictionarySize);
+    double originalColumnDictionarySize = originalColumnIndexSizes.get(StandardIndexes.DICTIONARY_ID).asDouble();
+    assertEquals(derivedColumnIndexSizes.get(StandardIndexes.DICTIONARY_ID).asDouble(), originalColumnDictionarySize);
+
     // Derived string column should have larger dictionary size than the original column
-    assertTrue(derivedStringColumnIndexSizes.get("dictionary").asDouble() > originalColumnDictionarySize);
+    assertTrue(
+        derivedStringColumnIndexSizes.get(StandardIndexes.DICTIONARY_ID).asDouble() > originalColumnDictionarySize);
+
     // Both derived columns should have smaller forward index size than the original column because of compression
-    double derivedColumnForwardIndexSize = derivedColumnIndexSizes.get("forward_index").asDouble();
-    assertTrue(derivedColumnForwardIndexSize < originalColumnIndexSizes.get("forward_index").asDouble());
-    assertEquals(derivedStringColumnIndexSizes.get("forward_index").asDouble(), derivedColumnForwardIndexSize);
+    double derivedColumnForwardIndexSize = derivedColumnIndexSizes.get(StandardIndexes.FORWARD_ID).asDouble();
+    assertTrue(derivedColumnForwardIndexSize < originalColumnIndexSizes.get(StandardIndexes.FORWARD_ID).asDouble());
+    assertEquals(derivedStringColumnIndexSizes.get(StandardIndexes.FORWARD_ID).asDouble(),
+        derivedColumnForwardIndexSize);
+
+    assertTrue(derivedRawStringColumnIndex.has(StandardIndexes.FORWARD_ID));
+    assertFalse(derivedRawStringColumnIndex.has(StandardIndexes.DICTIONARY_ID));
 
-    assertTrue(derivedRawStringColumnIndex.has("forward_index"));
-    assertFalse(derivedRawStringColumnIndex.has("dictionary"));
+    assertTrue(derivedRawMVIntColumnIndex.has(StandardIndexes.FORWARD_ID));
+    assertFalse(derivedRawMVIntColumnIndex.has(StandardIndexes.DICTIONARY_ID));
 
-    assertTrue(derivedRawMVIntColumnIndex.has("forward_index"));
-    assertFalse(derivedRawMVIntColumnIndex.has("dictionary"));
+    assertTrue(derivedNullStringColumnIndex.has(StandardIndexes.NULL_VALUE_VECTOR_ID));
   }
 
   private void reloadWithMissingColumns()
@@ -1753,6 +1765,9 @@ private void testNewAddedColumns()
         + (useMultiStageQueryEngine() ? "X''" : "''");
     testQuery(pinotQuery, h2Query);
 
+    pinotQuery = "SELECT COUNT(*) FROM mytable WHERE NewAddedDerivedNullString = 'nil'";
+    testQuery(pinotQuery, h2Query);
+
     pinotQuery = "SELECT COUNT(*) FROM mytable WHERE NewAddedDerivedHoursSinceEpoch = 392232";
     h2Query = "SELECT COUNT(*) FROM mytable WHERE DaysSinceEpoch = 16343";
     testQuery(pinotQuery, h2Query);

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyFunctionEvaluator.java

@@ -58,8 +58,10 @@ public class GroovyFunctionEvaluator implements FunctionEvaluator {
   private final int _numArguments;
   private final Binding _binding;
   private final Script _script;
+  private final String _expression;
 
   public GroovyFunctionEvaluator(String closure) {
+    _expression = closure;
     Matcher matcher = GROOVY_FUNCTION_PATTERN.matcher(closure);
     Preconditions.checkState(matcher.matches(), "Invalid transform expression: %s", closure);
     String arguments = matcher.group(ARGUMENTS_GROUP_NAME);
@@ -110,4 +112,9 @@ public Object evaluate(Object[] values) {
     }
     return _script.run();
   }
+
+  @Override
+  public String toString() {
+    return _expression;
+  }
 }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/InbuiltFunctionEvaluator.java

@@ -46,8 +46,10 @@ public class InbuiltFunctionEvaluator implements FunctionEvaluator {
   // Root of the execution tree
   private final ExecutableNode _rootNode;
   private final List<String> _arguments;
+  private final String _functionExpression;
 
   public InbuiltFunctionEvaluator(String functionExpression) {
+    _functionExpression = functionExpression;
     _arguments = new ArrayList<>();
     _rootNode = planExecution(RequestContextUtils.getExpression(functionExpression));
   }
@@ -118,6 +120,11 @@ public Object evaluate(Object[] values) {
     return _rootNode.execute(values);
   }
 
+  @Override
+  public String toString() {
+    return _functionExpression;
+  }
+
   private interface ExecutableNode {
 
     Object execute(GenericRow row);