diff --git a/core/src/main/java/org/apache/hop/core/Condition.java b/core/src/main/java/org/apache/hop/core/Condition.java index 738f83882b3..d3c5c223566 100644 --- a/core/src/main/java/org/apache/hop/core/Condition.java +++ b/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 { diff --git a/core/src/test/java/org/apache/hop/core/ConditionTest.java b/core/src/test/java/org/apache/hop/core/ConditionTest.java index 14054cc702b..9240a3c708d 100644 --- a/core/src/test/java/org/apache/hop/core/ConditionTest.java +++ b/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 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 rightMetas = new ArrayList<>(); + private final List 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> 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"));