feat(parkservices): preserve last descriptor fields; mark NaN feature indices in processSequentially (#414)

kaituo · web-flow · commit 1ab2a3c1bcbf · 2025-09-10T16:21:47.000-07:00
This PR
- `processSequentially`: before constructing each `AnomalyDescriptor`, scan the
  point for `Double.NaN` values and set `missingValues` to the indices of the
  missing features. This enables `processSequentially` to handle missing values.
- Preserve `currentInput` and `missingValues` during `ThresholdedRandomCutForest`
  serialization so the last processed descriptor survives save/restore. Enables
  downstream consumers to surface/impute per-feature missingness.

Testing
- Unit tests for TRCF round-trip serialization and NaN handling.
- Manual verification on sample streams.

Signed-off-by: kaituo &lt;kaituo@amazon.com&gt;
diff --git a/Java/README.md b/Java/README.md
@@ -157,7 +157,7 @@ vector data point, scores the data point, and then updates the model with this
 point. The program output appends a column of anomaly scores to the input:
 
 ```text
-$ java -cp core/target/randomcutforest-core-4.3.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner < ../example-data/rcf-paper.csv > example_output.csv
+$ java -cp core/target/randomcutforest-core-4.4.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner < ../example-data/rcf-paper.csv > example_output.csv
 $ tail example_output.csv
 -5.0029,0.0170,-0.0057,0.8129401629464965
 -4.9975,-0.0102,-0.0065,0.6591046054520615
@@ -176,8 +176,8 @@ read additional usage instructions, including options for setting model
 hyperparameters, using the `--help` flag:
 
 ```text
-$ java -cp core/target/randomcutforest-core-4.3.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner --help
-Usage: java -cp target/random-cut-forest-4.3.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner [options] < input_file > output_file
+$ java -cp core/target/randomcutforest-core-4.4.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner --help
+Usage: java -cp target/random-cut-forest-4.4.0.jar com.amazon.randomcutforest.runner.AnomalyScoreRunner [options] < input_file > output_file
 
 Compute scalar anomaly scores from the input rows and append them to the output rows.
 
@@ -239,14 +239,14 @@ framework. Build an executable jar containing the benchmark code by running
 To invoke the full benchmark suite:
 
 ```text
-% java -jar benchmark/target/randomcutforest-benchmark-4.3.0-jar-with-dependencies.jar
+% java -jar benchmark/target/randomcutforest-benchmark-4.4.0-jar-with-dependencies.jar
 ```
 
 The full benchmark suite takes a long time to run. You can also pass a regex at the command-line, then only matching
 benchmark methods will be executed.
 
 ```text
-% java -jar benchmark/target/randomcutforest-benchmark-4.3.0-jar-with-dependencies.jar RandomCutForestBenchmark\.updateAndGetAnomalyScore
+% java -jar benchmark/target/randomcutforest-benchmark-4.4.0-jar-with-dependencies.jar RandomCutForestBenchmark\.updateAndGetAnomalyScore
 ```
 
 [rcf-paper]: http://proceedings.mlr.press/v48/guha16.pdf
diff --git a/Java/benchmark/pom.xml b/Java/benchmark/pom.xml
@@ -6,7 +6,7 @@
   <parent>
     <groupId>software.amazon.randomcutforest</groupId>
     <artifactId>randomcutforest-parent</artifactId>
-    <version>4.3.0</version>
+    <version>4.4.0</version>
   </parent>
 
   <artifactId>randomcutforest-benchmark</artifactId>
diff --git a/Java/core/pom.xml b/Java/core/pom.xml
@@ -6,7 +6,7 @@
   <parent>
     <groupId>software.amazon.randomcutforest</groupId>
     <artifactId>randomcutforest-parent</artifactId>
-    <version>4.3.0</version>
+    <version>4.4.0</version>
   </parent>
 
   <artifactId>randomcutforest-core</artifactId>
diff --git a/Java/examples/pom.xml b/Java/examples/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>software.amazon.randomcutforest</groupId>
         <artifactId>randomcutforest-parent</artifactId>
-        <version>4.3.0</version>
+        <version>4.4.0</version>
     </parent>
 
     <artifactId>randomcutforest-examples</artifactId>
diff --git a/Java/parkservices/pom.xml b/Java/parkservices/pom.xml
@@ -6,7 +6,7 @@
   <parent>
     <groupId>software.amazon.randomcutforest</groupId>
     <artifactId>randomcutforest-parent</artifactId>
-    <version>4.3.0</version>
+    <version>4.4.0</version>
   </parent>
 
   <artifactId>randomcutforest-parkservices</artifactId>
diff --git a/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/PredictorCorrector.java b/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/PredictorCorrector.java
@@ -730,6 +730,7 @@ protected void saveScores(ScoringStrategy strategy, int choice, double[] scoreVe
     protected <P extends AnomalyDescriptor> P detect(P result, RCFComputeDescriptor lastSignificantDescriptor,
             RandomCutForest forest) {
         if (result.getRCFPoint() == null) {
+            lastDescriptor = result.copyOf();
             return result;
         }
         float[] point = result.getRCFPoint();
@@ -747,6 +748,7 @@ protected <P extends AnomalyDescriptor> P detect(P result, RCFComputeDescriptor
 
         // we will not have zero scores affect any thresholding
         if (score == 0) {
+            lastDescriptor = result.copyOf();
             return result;
         }
 
diff --git a/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/ThresholdedRandomCutForest.java b/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/ThresholdedRandomCutForest.java
@@ -318,8 +318,11 @@ public List<AnomalyDescriptor> processSequentially(double[][] data, Function<Ano
      * can have additional benefits. At the moment the operation does not support
      * external timestamps.
      *
+     *
+     *
      * @param data       a vectors of vectors (each of which has to have the same
-     *                   inputLength)
+     *                   inputLength). Mising values are represented by Double.NaN
+     *                   in a vector.
      * @param timestamps a vector of timestamps (in the same order as the data, has
      *                   to be same length as data, and ascending)
      * @param filter     a condition to drop desriptor (recommended filter:
@@ -368,6 +371,11 @@ public List<AnomalyDescriptor> processSequentially(double[][] data, long[] times
                     checkArgument(point != null, " data should not be null ");
                     checkArgument(point.length == length, " nonuniform lengths ");
                     AnomalyDescriptor description = new AnomalyDescriptor(point, timestamp);
+                    // check missing values in point.
+                    int[] missingValues = generateMissingIndicesArray(point);
+                    if (missingValues != null) {
+                        description.setMissingValues(missingValues);
+                    }
                     augment(description);
                     if (saveDescriptor(description)) {
                         lastAnomalyDescriptor = description.copyOf();
@@ -390,6 +398,20 @@ public List<AnomalyDescriptor> processSequentially(double[][] data) {
         return processSequentially(data, x -> x.getAnomalyGrade() > 0);
     }
 
+    private int[] generateMissingIndicesArray(double[] point) {
+        List<Integer> intArray = new ArrayList<>();
+        for (int i = 0; i < point.length; i++) {
+            if (Double.isNaN(point[i])) {
+                intArray.add(i);
+            }
+        }
+        // Return null if the array is empty
+        if (intArray.size() == 0) {
+            return null;
+        }
+        return intArray.stream().mapToInt(Integer::intValue).toArray();
+    }
+
     /**
      * a function that extrapolates the data seen by the ThresholdedRCF model, and
      * uses the transformations allowed (as opposed to just using RCFs). The
diff --git a/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/state/returntypes/ComputeDescriptorMapper.java b/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/state/returntypes/ComputeDescriptorMapper.java
@@ -34,7 +34,7 @@ public class ComputeDescriptorMapper implements IStateMapper<RCFComputeDescripto
     @Override
     public RCFComputeDescriptor toModel(ComputeDescriptorState state, long seed) {
 
-        RCFComputeDescriptor descriptor = new RCFComputeDescriptor(null, 0L);
+        RCFComputeDescriptor descriptor = new RCFComputeDescriptor(state.getCurrentInput(), state.getInputTimeStamp());
         descriptor.setRCFScore(state.getScore());
         descriptor.setInternalTimeStamp(state.getInternalTimeStamp());
         descriptor.setAttribution(new DiVectorMapper().toModel(state.getAttribution()));
@@ -72,6 +72,8 @@ public ComputeDescriptorState toState(RCFComputeDescriptor descriptor) {
         state.setAnomalyGrade(descriptor.getAnomalyGrade());
         state.setThreshold(descriptor.getThreshold());
         state.setCorrectionMode(descriptor.getCorrectionMode().name());
+        state.setInputTimeStamp(descriptor.getInputTimestamp());
+        state.setCurrentInput(descriptor.getCurrentInput());
         return state;
     }
 
diff --git a/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/state/returntypes/ComputeDescriptorState.java b/Java/parkservices/src/main/java/com/amazon/randomcutforest/parkservices/state/returntypes/ComputeDescriptorState.java
@@ -23,7 +23,7 @@
 
 @Data
 public class ComputeDescriptorState implements Serializable {
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 2L;
 
     private long internalTimeStamp;
     private double score;
@@ -42,4 +42,6 @@ public class ComputeDescriptorState implements Serializable {
     private double threshold;
     private double anomalyGrade;
     private String correctionMode;
+    private long inputTimeStamp;
+    private double[] currentInput;
 }
diff --git a/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/ThresholdedRandomCutForestTest.java b/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/ThresholdedRandomCutForestTest.java
@@ -672,4 +672,38 @@ void psqEmptyDataReturnsEmpty() {
         assertTrue(out2.isEmpty());
     }
 
+    @Test
+    void testProcessSequentiallyWithMissingValues() {
+        ThresholdedRandomCutForest f = ThresholdedRandomCutForest.builder().dimensions(2).shingleSize(1).build();
+
+        double[][] data = { { 1.0, Double.NaN }, { 3.0, 4.0 }, { Double.NaN, 5.0 }, { Double.NaN, Double.NaN } };
+        long[] stamps = { 10L, 20L, 30L, 40L };
+
+        List<AnomalyDescriptor> descriptors = f.processSequentially(data, stamps, d -> true);
+
+        assertEquals(4, descriptors.size());
+
+        AnomalyDescriptor first = descriptors.get(0);
+        assertEquals(10L, first.getInputTimestamp());
+        assertArrayEquals(new double[] { 1.0, Double.NaN }, first.getCurrentInput());
+        assertNotNull(first.getMissingValues());
+        assertArrayEquals(new int[] { 1 }, first.getMissingValues());
+
+        AnomalyDescriptor second = descriptors.get(1);
+        assertEquals(20L, second.getInputTimestamp());
+        assertArrayEquals(new double[] { 3.0, 4.0 }, second.getCurrentInput());
+        assertNull(second.getMissingValues());
+
+        AnomalyDescriptor third = descriptors.get(2);
+        assertEquals(30L, third.getInputTimestamp());
+        assertArrayEquals(new double[] { Double.NaN, 5.0 }, third.getCurrentInput());
+        assertNotNull(third.getMissingValues());
+        assertArrayEquals(new int[] { 0 }, third.getMissingValues());
+
+        AnomalyDescriptor fourth = descriptors.get(3);
+        assertEquals(40L, fourth.getInputTimestamp());
+        assertArrayEquals(new double[] { Double.NaN, Double.NaN }, fourth.getCurrentInput());
+        assertNotNull(fourth.getMissingValues());
+        assertArrayEquals(new int[] { 0, 1 }, fourth.getMissingValues());
+    }
 }
diff --git a/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/TransformTest.java b/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/TransformTest.java
@@ -42,10 +42,20 @@ public void AnomalyTest(TransformMethod method) {
             int numberOfTrees = 30 + rng.nextInt(20);
             int outputAfter = 32 + rng.nextInt(50);
             // shingleSize 1 is not recommended for complicated input
-            int shingleSize = 2 + rng.nextInt(15);
+            // The test sets alertOnce(true) to suppress cascades after a single injected spike.
+            // Current suppression only triggers for overlapping shingles: gap < shingleSize
+            // (in PredictorCorrector.detect), so “late” follow-on spikes (gap ≥ shingleSize)
+            // still raise grades.
+            // If shingleSize is small (e.g., 2), gap is small, anomaly would not be suppressed,
+            // may cause totalcount > numTrials.
+            int shingleSize = 3 + rng.nextInt(15);
             int baseDimensions = 1 + rng.nextInt(5);
             int dimensions = baseDimensions * shingleSize;
             long forestSeed = rng.nextLong();
+            System.out.println(" forestSeed " + forestSeed + " method " + method + " seed " + seed + " outputAfter "
+                    + outputAfter + " shingleSize " + shingleSize + " baseDimensions " + baseDimensions + " dimensions "
+                    + dimensions + " numberOfTrees " + numberOfTrees + " rng " + rng + " i " + i + " shingleSize "
+                    + shingleSize + " rng.nextLong() ");
             ThresholdedRandomCutForest first = new ThresholdedRandomCutForest.Builder<>().dimensions(dimensions)
                     .numberOfTrees(numberOfTrees).randomSeed(forestSeed).outputAfter(outputAfter).alertOnce(true)
                     .transformMethod(method).internalShinglingEnabled(true).shingleSize(shingleSize).build();
diff --git a/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/state/ThresholdedRandomCutForestMapperTest.java b/Java/parkservices/src/test/java/com/amazon/randomcutforest/parkservices/state/ThresholdedRandomCutForestMapperTest.java
@@ -19,6 +19,8 @@
 import static com.amazon.randomcutforest.preprocessor.Preprocessor.shiftLeft;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 import java.util.Random;
 import java.util.stream.Stream;
@@ -36,6 +38,7 @@
 import com.amazon.randomcutforest.config.TransformMethod;
 import com.amazon.randomcutforest.parkservices.AnomalyDescriptor;
 import com.amazon.randomcutforest.parkservices.ThresholdedRandomCutForest;
+import com.amazon.randomcutforest.parkservices.returntypes.RCFComputeDescriptor;
 import com.amazon.randomcutforest.returntypes.TimedRangeVector;
 import com.amazon.randomcutforest.state.RandomCutForestMapper;
 import com.amazon.randomcutforest.testutils.MultiDimDataWithKey;
@@ -707,6 +710,75 @@ public void testRoundTripImpute(TransformMethod transformMethod, ImputationMetho
         }
     }
 
+    @Test
+    public void testRoundTripLastDescriptor() {
+        int dimensions = 2;
+        long seed = new Random().nextLong();
+
+        ThresholdedRandomCutForest trcf = new ThresholdedRandomCutForest.Builder<>().dimensions(dimensions)
+                .randomSeed(seed).build();
+
+        double[] point = { 1.0, Double.NaN };
+        long timestamp = 123L;
+
+        trcf.process(point, timestamp, new int[] { 1 });
+
+        ThresholdedRandomCutForestMapper mapper = new ThresholdedRandomCutForestMapper();
+        ThresholdedRandomCutForestState state = mapper.toState(trcf);
+        ThresholdedRandomCutForest deserializedTrcf = mapper.toModel(state);
+
+        RCFComputeDescriptor deserializedDescriptor = deserializedTrcf.getPredictorCorrector().getLastDescriptor();
+        assertNotNull(deserializedDescriptor);
+        assertArrayEquals(point, deserializedDescriptor.getCurrentInput());
+        assertEquals(timestamp, deserializedDescriptor.getInputTimestamp());
+    }
+
+    @Test
+    public void testRoundTripProcessSequentially() {
+        int dimensions = 2;
+        long seed = new Random().nextLong();
+        ThresholdedRandomCutForestMapper mapper = new ThresholdedRandomCutForestMapper();
+
+        // Case 1: Zero input
+        ThresholdedRandomCutForest emptyTrcf = new ThresholdedRandomCutForest.Builder<>().dimensions(dimensions)
+                .randomSeed(seed).build();
+        ThresholdedRandomCutForestState emptyState = mapper.toState(emptyTrcf);
+        ThresholdedRandomCutForest deserializedEmptyTrcf = mapper.toModel(emptyState);
+        assertNull(deserializedEmptyTrcf.getPredictorCorrector().getLastDescriptor());
+
+        // Case 2: 100 inputs
+        ThresholdedRandomCutForest trcf = new ThresholdedRandomCutForest.Builder<>().dimensions(dimensions)
+                .randomSeed(seed).build();
+
+        int numPoints = 100;
+        double[][] data = new double[numPoints][dimensions];
+        long[] timestamps = new long[numPoints];
+        Random random = new Random(seed);
+
+        for (int i = 0; i < numPoints; i++) {
+            for (int j = 0; j < dimensions; j++) {
+                data[i][j] = random.nextDouble();
+            }
+            timestamps[i] = i * 100L;
+        }
+        // Set the last point to have a missing value
+        data[numPoints - 1][dimensions - 1] = Double.NaN;
+
+        trcf.processSequentially(data, timestamps, d -> true);
+
+        ThresholdedRandomCutForestState state = mapper.toState(trcf);
+        ThresholdedRandomCutForest deserializedTrcf = mapper.toModel(state);
+
+        RCFComputeDescriptor deserializedDescriptor = deserializedTrcf.getPredictorCorrector().getLastDescriptor();
+        assertNotNull(deserializedDescriptor);
+
+        double[] lastPoint = data[numPoints - 1];
+        long lastTimestamp = timestamps[numPoints - 1];
+
+        assertArrayEquals(lastPoint, deserializedDescriptor.getCurrentInput());
+        assertEquals(lastTimestamp, deserializedDescriptor.getInputTimestamp());
+    }
+
     static Stream<Arguments> args() {
         return transformMethodStream().flatMap(
                 classParameter -> imputationMethod().map(testParameter -> Arguments.of(classParameter, testParameter)));
diff --git a/Java/pom.xml b/Java/pom.xml
@@ -4,7 +4,7 @@
 
     <groupId>software.amazon.randomcutforest</groupId>
     <artifactId>randomcutforest-parent</artifactId>
-    <version>4.3.0</version>
+    <version>4.4.0</version>
     <packaging>pom</packaging>
 
     <name>software.amazon.randomcutforest:randomcutforest</name>
diff --git a/Java/serialization/pom.xml b/Java/serialization/pom.xml
@@ -7,7 +7,7 @@
     <parent>
         <groupId>software.amazon.randomcutforest</groupId>
         <artifactId>randomcutforest-parent</artifactId>
-        <version>4.3.0</version>
+        <version>4.4.0</version>
     </parent>
 
     <artifactId>randomcutforest-serialization</artifactId>
diff --git a/Java/testutils/pom.xml b/Java/testutils/pom.xml
@@ -4,7 +4,7 @@
   <parent>
     <artifactId>randomcutforest-parent</artifactId>
     <groupId>software.amazon.randomcutforest</groupId>
-    <version>4.3.0</version>
+    <version>4.4.0</version>
   </parent>
 
   <artifactId>randomcutforest-testutils</artifactId>

Original file line number	Diff line number	Diff line change
`@@ -730,6 +730,7 @@ protected void saveScores(ScoringStrategy strategy, int choice, double[] scoreVe`
`730`	`730`	`protected <P extends AnomalyDescriptor> P detect(P result, RCFComputeDescriptor lastSignificantDescriptor,`
`731`	`731`	`RandomCutForest forest) {`
`732`	`732`	`if (result.getRCFPoint() == null) {`
	`733`	`+ lastDescriptor = result.copyOf();`
`733`	`734`	`return result;`
`734`	`735`	`}`
`735`	`736`	`float[] point = result.getRCFPoint();`
`@@ -747,6 +748,7 @@ protected <P extends AnomalyDescriptor> P detect(P result, RCFComputeDescriptor`
`747`	`748`
`748`	`749`	`// we will not have zero scores affect any thresholding`
`749`	`750`	`if (score == 0) {`
	`751`	`+ lastDescriptor = result.copyOf();`
`750`	`752`	`return result;`
`751`	`753`	`}`
`752`	`754`