From 3c68ff2737cafa8c821ab6275c7f6e72c783de51 Mon Sep 17 00:00:00 2001 From: Saurabh Rusia Date: Fri, 2 Oct 2026 20:50:46 +0530 Subject: [PATCH 1/2] DLPX-99491 Hop 2.15: delimited file masking ~3x slower because Condition.evaluate builds a ValueMeta per row PR URL: https://www.github.com/delphix/hop/pull/16 --- .../java/org/apache/hop/core/Condition.java | 96 +++++++++++- .../org/apache/hop/core/ConditionTest.java | 142 ++++++++++++++++++ 2 files changed, 236 insertions(+), 2 deletions(-) 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..c3f20bf0840 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,16 @@ public static final class CValue { @HopMetadataProperty(key = "mask", isExcludedFromInjection = true) private String mask; + /* + * DLPX-99491: both createValueMeta() and createValueData() build ValueMeta objects whose + * constructor reads two system properties, and evaluate() calls them per row. 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 +872,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..6b815f0b7c6 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,23 @@ 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 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 +96,135 @@ 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})); + } + + /** + * 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 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( + () -> { + readySetGo.await(); + for (int n = 0; n < 500; n++) { + assertNotNull(rightValue.getCachedValueMeta()); + assertEquals(Long.valueOf(10000L), rightValue.getCachedValueData()); + } + return null; + })); + } + // Release every thread onto a cold cache at the same moment. + 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")); From a1e7e9b6d32a211f44ca203cbc2b031c900b1e28 Mon Sep 17 00:00:00 2001 From: Saurabh Rusia Date: Mon, 5 Oct 2026 23:21:22 +0530 Subject: [PATCH 2/2] Review comments fixes --- .../java/org/apache/hop/core/Condition.java | 6 ++-- .../org/apache/hop/core/ConditionTest.java | 30 ++++++++++++++++++- 2 files changed, 31 insertions(+), 5 deletions(-) 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 c3f20bf0840..d3c5c223566 100644 --- a/core/src/main/java/org/apache/hop/core/Condition.java +++ b/core/src/main/java/org/apache/hop/core/Condition.java @@ -797,10 +797,8 @@ public static final class CValue { private String mask; /* - * DLPX-99491: both createValueMeta() and createValueData() build ValueMeta objects whose - * constructor reads two system properties, and evaluate() calls them per row. 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. + * 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) 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 6b815f0b7c6..9240a3c708d 100644 --- a/core/src/test/java/org/apache/hop/core/ConditionTest.java +++ b/core/src/test/java/org/apache/hop/core/ConditionTest.java @@ -32,6 +32,7 @@ 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; @@ -143,6 +144,29 @@ public void testConstantRightValueIsRebuiltAfterMutation() throws Exception { 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()); } /** @@ -198,6 +222,7 @@ public void testCachedRightValueTolerateConcurrentColdAccess() throws Exception 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<>(); @@ -206,6 +231,7 @@ public void testCachedRightValueTolerateConcurrentColdAccess() throws Exception results.add( pool.submit( () -> { + allArrived.countDown(); readySetGo.await(); for (int n = 0; n < 500; n++) { assertNotNull(rightValue.getCachedValueMeta()); @@ -214,7 +240,9 @@ public void testCachedRightValueTolerateConcurrentColdAccess() throws Exception return null; })); } - // Release every thread onto a cold cache at the same moment. + // 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.