DLPX-99491 Hop 2.15: delimited file masking (with header/trailer) ~3x slower because Condition.evaluate builds a ValueMeta per row

core/src/main/java/org/apache/hop/core/Condition.java

@@ -29,7 +29,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.regex.Pattern;
+import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.commons.lang.StringUtils;
@@ -356,12 +358,12 @@ public boolean evaluate(IRowMeta rowMeta, Object[] r) {
 
         // Get field index: right value
         //
-        IValueMeta fieldMeta2 = rightValue != null ? rightValue.createValueMeta() : null;
+        IValueMeta fieldMeta2 = rightValue != null ? rightValue.getCachedValueMeta() : null;
         // Old metadata contains a right value block without name, type and so on.  This means: no
         // value
         // Removed the name check, old fixed values do not contain a name element causing regression
         Object field2 =
-            rightValue != null && rightFieldIndex == -2 ? rightValue.createValueData() : null;
+            rightValue != null && rightFieldIndex == -2 ? rightValue.getCachedValueData() : null;
         if (field2 == null && rightFieldIndex >= 0) {
           fieldMeta2 = rowMeta.getValueMeta(rightFieldIndex);
           field2 = r[rightFieldIndex];
@@ -794,6 +796,14 @@ public static final class CValue {
     @HopMetadataProperty(key = "mask", isExcludedFromInjection = true)
     private String mask;
 
+    /*
+     * Cache the result alongside the field values it came from, so any later mutation rebuilds it.
+     * Volatile because one Condition is shared by every copy of a transform.
+     */
+    @Getter(AccessLevel.NONE)
+    @Setter(AccessLevel.NONE)
+    private volatile CachedRightValue cachedRightValue;
+
     public CValue() {}
 
     public CValue(CValue c) {
@@ -860,6 +870,86 @@ public Object createValueData() throws HopException {
 
       return stringValueMeta.convertDataUsingConversionMetaData(val.getValueData());
     }
+
+    /**
+     * As {@link #createValueMeta()} but built once and shared until a field changes. Intended for
+     * the per-row evaluate() path; callers that may modify the result must keep using
+     * createValueMeta().
+     *
+     * @return The shared value metadata describing the right value of the condition
+     * @throws HopPluginException
+     */
+    IValueMeta getCachedValueMeta() throws HopPluginException {
+      return cached().valueMeta;
+    }
+
+    /**
+     * As {@link #createValueData()} but built once and shared until a field changes. Intended for
+     * the per-row evaluate() path; callers that may modify the result must keep using
+     * createValueData().
+     *
+     * @return The shared converted right value, which may be null
+     * @throws HopException
+     */
+    Object getCachedValueData() throws HopException {
+      CachedRightValue current = cached();
+      if (!current.valueDataBuilt) {
+        current.valueData = createValueData();
+        // Set last: a thread seeing this flag is guaranteed to see the value above.
+        current.valueDataBuilt = true;
+      }
+      return current.valueData;
+    }
+
+    private CachedRightValue cached() throws HopPluginException {
+      CachedRightValue current = cachedRightValue;
+      if (current == null || !current.describes(this)) {
+        current = new CachedRightValue(this, createValueMeta());
+        cachedRightValue = current;
+      }
+      return current;
+    }
+
+    /**
+     * A cached right value together with the field values it was derived from, so that mutating a
+     * CValue - through its setters or through metadata injection - discards it rather than leaving
+     * stale metadata behind. The value data is built on demand because evaluate() only needs it
+     * when the condition has no right field.
+     */
+    private static final class CachedRightValue {
+      private final String name;
+      private final String type;
+      private final String text;
+      private final int length;
+      private final int precision;
+      private final boolean nullValue;
+      private final String mask;
+
+      private final IValueMeta valueMeta;
+      private volatile Object valueData;
+      private volatile boolean valueDataBuilt;
+
+      private CachedRightValue(CValue value, IValueMeta valueMeta) {
+        this.name = value.name;
+        this.type = value.type;
+        this.text = value.text;
+        this.length = value.length;
+        this.precision = value.precision;
+        this.nullValue = value.nullValue;
+        this.mask = value.mask;
+        this.valueMeta = valueMeta;
+      }
+
+      private boolean describes(CValue value) {
+        return length == value.length
+            && precision == value.precision
+            && nullValue == value.nullValue
+            && Objects.equals(name, value.name)
+            && Objects.equals(type, value.type)
+            && Objects.equals(text, value.text)
+            && Objects.equals(mask, value.mask);
+      }
+    }
   }
 
   public enum Operator implements IEnumHasCode {

core/src/test/java/org/apache/hop/core/ConditionTest.java

@@ -18,10 +18,24 @@
 package org.apache.hop.core;
 
 import static org.apache.hop.core.Condition.Function;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.hop.core.exception.HopValueException;
 import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
 import org.apache.hop.core.row.RowMeta;
 import org.apache.hop.core.row.ValueMetaAndData;
 import org.apache.hop.core.row.value.ValueMetaInteger;
@@ -83,6 +97,162 @@ public void testNullLessThanNumberEvaluatesAsFalse() throws Exception {
     assertFalse(condition.evaluate(rowMeta1, new Object[] {null, "test"}));
   }
 
+  @Test
+  public void testConstantRightValueStaysCorrectAcrossRows() throws Exception {
+    IRowMeta rowMeta = new RowMeta();
+    rowMeta.addValueMeta(new ValueMetaInteger("name1"));
+
+    ValueMetaAndData rightExact = new ValueMetaAndData(new ValueMetaInteger("name1"), 10000L);
+    Condition condition = new Condition("name1", Function.SMALLER, null, rightExact);
+
+    // The right value is built once and then shared, so repeated evaluation must not drift.
+    assertTrue(condition.evaluate(rowMeta, new Object[] {5000L}));
+    assertFalse(condition.evaluate(rowMeta, new Object[] {15000L}));
+    assertTrue(condition.evaluate(rowMeta, new Object[] {9999L}));
+    assertFalse(condition.evaluate(rowMeta, new Object[] {10000L}));
+  }
+
+  @Test
+  public void testConstantRightValueIsBuiltOnlyOnce() throws Exception {
+    IRowMeta rowMeta = new RowMeta();
+    rowMeta.addValueMeta(new ValueMetaInteger("name1"));
+
+    ValueMetaAndData rightExact = new ValueMetaAndData(new ValueMetaInteger("name1"), 10000L);
+    Condition condition = new Condition("name1", Function.SMALLER, null, rightExact);
+    assertTrue(condition.evaluate(rowMeta, new Object[] {5000L}));
+
+    Condition.CValue rightValue = condition.getRightValue();
+    assertSame(rightValue.getCachedValueMeta(), rightValue.getCachedValueMeta());
+    assertSame(rightValue.getCachedValueData(), rightValue.getCachedValueData());
+    // createValueMeta() must keep handing fresh objects to callers that modify the result.
+    assertNotSame(rightValue.getCachedValueMeta(), rightValue.createValueMeta());
+  }
+
+  @Test
+  public void testConstantRightValueIsRebuiltAfterMutation() throws Exception {
+    IRowMeta rowMeta = new RowMeta();
+    rowMeta.addValueMeta(new ValueMetaInteger("name1"));
+
+    ValueMetaAndData rightExact = new ValueMetaAndData(new ValueMetaInteger("name1"), 10000L);
+    Condition condition = new Condition("name1", Function.SMALLER, null, rightExact);
+    assertTrue(condition.evaluate(rowMeta, new Object[] {5000L}));
+
+    // Mutating the right value must discard the cache rather than keep comparing against 10000.
+    condition.getRightValue().setText("1000");
+    assertFalse(condition.evaluate(rowMeta, new Object[] {5000L}));
+    assertTrue(condition.evaluate(rowMeta, new Object[] {999L}));
+
+    condition.getRightValue().setNullValue(true);
+    assertFalse(condition.evaluate(rowMeta, new Object[] {999L}));
+
+    /*
+     * Changing name, length, precision or mask does not alter whether 5000 < 10000 - so assert
+     * that each one on its own replaces the cached instance.
+     */
+    Condition.CValue rightValue = condition.getRightValue();
+    assertCacheRebuilt("name", rightValue, value -> value.setName("other"));
+    assertCacheRebuilt("type", rightValue, value -> value.setType("String"));
+    assertCacheRebuilt("text", rightValue, value -> value.setText("77"));
+    assertCacheRebuilt("length", rightValue, value -> value.setLength(10));
+    assertCacheRebuilt("precision", rightValue, value -> value.setPrecision(2));
+    assertCacheRebuilt("nullValue", rightValue, value -> value.setNullValue(false));
+    assertCacheRebuilt("mask", rightValue, value -> value.setMask("#.##"));
+  }
+
+  private void assertCacheRebuilt(
+      String field, Condition.CValue value, Consumer<Condition.CValue> mutation) throws Exception {
+    IValueMeta cachedBefore = value.getCachedValueMeta();
+    mutation.accept(value);
+    assertNotSame(
+        "changing " + field + " must invalidate the cached right value",
+        cachedBefore,
+        value.getCachedValueMeta());
+  }
+
+  /**
+   * Pins the performance contract itself: evaluate() must hand the same right-side instances to
+   * every row rather than rebuilding them. Asserting on the cache accessors alone is not enough,
+   * because they would keep returning a stable instance even if evaluate() went back to calling
+   * createValueMeta() / createValueData() per row.
+   */
+  @Test
+  public void testEvaluateReusesTheRightValueAcrossRows() throws Exception {
+    CapturingValueMeta leftMeta = new CapturingValueMeta("name1");
+    IRowMeta rowMeta = new RowMeta();
+    rowMeta.addValueMeta(leftMeta);
+
+    ValueMetaAndData rightExact = new ValueMetaAndData(new ValueMetaInteger("name1"), 10000L);
+    Condition condition = new Condition("name1", Function.SMALLER, null, rightExact);
+
+    assertTrue(condition.evaluate(rowMeta, new Object[] {5000L}));
+    assertTrue(condition.evaluate(rowMeta, new Object[] {6000L}));
+
+    assertEquals(2, leftMeta.rightMetas.size());
+    assertSame(leftMeta.rightMetas.get(0), leftMeta.rightMetas.get(1));
+    assertSame(leftMeta.rightData.get(0), leftMeta.rightData.get(1));
+  }
+
+  /** Records the right-side metadata and data that evaluate() passes into the left value. */
+  private static final class CapturingValueMeta extends ValueMetaInteger {
+    private final List<IValueMeta> rightMetas = new ArrayList<>();
+    private final List<Object> rightData = new ArrayList<>();
+
+    private CapturingValueMeta(String name) {
+      super(name);
+    }
+
+    @Override
+    public int compare(Object data1, IValueMeta meta2, Object data2) throws HopValueException {
+      rightMetas.add(meta2);
+      rightData.add(data2);
+      return super.compare(data1, meta2, data2);
+    }
+  }
+
+  /**
+   * One Condition is shared by every copy of a transform, so the cached right value has to tolerate
+   * concurrent first access. This exercises the accessors directly rather than evaluate(), which
+   * writes leftFieldIndex, rightFieldIndex, rightValue, inList and rightString on the shared
+   * instance and so is not thread safe regardless of any caching.
+   */
+  @Test
+  public void testCachedRightValueTolerateConcurrentColdAccess() throws Exception {
+    ValueMetaAndData rightExact = new ValueMetaAndData(new ValueMetaInteger("name1"), 10000L);
+    Condition condition = new Condition("name1", Function.SMALLER, null, rightExact);
+    Condition.CValue rightValue = condition.getRightValue();
+
+    int threads = 8;
+    CountDownLatch allArrived = new CountDownLatch(threads);
+    CountDownLatch readySetGo = new CountDownLatch(1);
+    ExecutorService pool = Executors.newFixedThreadPool(threads);
+    List<Future<?>> results = new ArrayList<>();
+    try {
+      for (int i = 0; i < threads; i++) {
+        results.add(
+            pool.submit(
+                () -> {
+                  allArrived.countDown();
+                  readySetGo.await();
+                  for (int n = 0; n < 500; n++) {
+                    assertNotNull(rightValue.getCachedValueMeta());
+                    assertEquals(Long.valueOf(10000L), rightValue.getCachedValueData());
+                  }
+                  return null;
+                }));
+      }
+      // Pool threads start lazily, so wait until every worker is parked on the start gate before
+      // opening it. Otherwise a late starter would find the cache already warm.
+      assertTrue("all workers should reach the start gate", allArrived.await(30, TimeUnit.SECONDS));
+      readySetGo.countDown();
+      for (Future<?> result : results) {
+        // Rethrows any assertion failure or exception raised on a worker thread.
+        result.get(30, TimeUnit.SECONDS);
+      }
+    } finally {
+      pool.shutdownNow();
+    }
+  }
+
   @Test
   public void testSerialization() throws Exception {
     Document document = XmlHandler.loadXmlFile(getClass().getResourceAsStream("/condition.xml"));