diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
index 3e76ce64316..f26cf595853 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala
@@ -40,6 +40,9 @@ trait EndChannelHandler {
     val portId = dp.inputGateway.getChannel(channelId).getPortId
     dp.inputManager.getPort(portId).completed = true
     dp.inputManager.initBatch(channelId, Array.empty)
+    // Before the operator finishes the port, so one that received no rows on
+    // it can still say what its columns were.
+    dp.executor.inputSchemas(portId.id) = dp.inputManager.getPort(portId).schema
     try {
       // The executor finishes: a setting that refers to loop variables must have them by now.
       dp.bindStateReferences()
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala
index d768b93c7d5..e8d9c50c6bf 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandlerSpec.scala
@@ -236,6 +236,18 @@ class EndChannelHandlerSpec extends AnyFlatSpec {
     assert(executor.finishOutputPorts.toList == List(currentPortId.id))
   }
 
+  it should "give the operator the finished port's declared schema before it finishes the port" in {
+    // A HashJoin probe worker that received no build rows names its unmatched right rows from
+    // this, so it has to be in place when the operator is asked to finish, not after.
+    val executor = new RecordingExecutor()
+    val fixture = new Fixture(executor, otherInputPortCompleted = false)
+
+    fixture.endChannel()
+
+    assert(executor.schemasAtFinish.toList == List(Map(currentPortId.id -> schema)))
+    assert(!executor.inputSchemas.contains(otherPortId.id))
+  }
+
   it should "emit the boundary state the operator produced on finish, with no loop envelope" in {
     val state = State(Map("boundary" -> "end"))
     val fixture = new Fixture(new RecordingExecutor(onFinishState = _ => Some(state)))
@@ -464,8 +476,12 @@ object EndChannelHandlerSpec {
     /** Ports handed to `onFinish`, i.e. what `onFinishMultiPort` forwarded. */
     val finishOutputPorts: ArrayBuffer[Int] = ArrayBuffer()
 
+    /** `inputSchemas` as the first finish callback saw it. */
+    val schemasAtFinish: ArrayBuffer[Map[Int, Schema]] = ArrayBuffer()
+
     override def produceStateOnFinish(port: Int): Option[State] = {
       statePorts += port
+      schemasAtFinish += inputSchemas.toMap
       onFinishState(port)
     }
 
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala
index e5b5f63818b..a37a566c163 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala
@@ -21,9 +21,11 @@ package org.apache.texera.amber.core.executor
 
 import com.google.common.collect.MapMaker
 import org.apache.texera.amber.core.state.State
-import org.apache.texera.amber.core.tuple.{Tuple, TupleLike}
+import org.apache.texera.amber.core.tuple.{Schema, Tuple, TupleLike}
 import org.apache.texera.amber.core.workflow.PortIdentity
 
+import scala.collection.mutable
+
 /**
   * Inside a loop block a setting may refer to a loop variable (`$i`): the worker writes each
   * iteration's value into the settings object the executor holds, in place (`registerState`). For
@@ -91,6 +93,17 @@ trait OperatorExecutor {
     */
   final def bindStateReferences(): Unit =
     Option(OperatorExecutor.bindings.get(this)).foreach(_.bind())
+
+  /**
+    * What each finished input port was declared to carry, keyed by port index
+    * and written by the runtime before `onFinish` for that port.
+    *
+    * The tuples say this too, so an operator only needs to ask when there were
+    * none: a port can finish having carried no rows, and its schema is then the
+    * only record of what its columns were.
+    */
+  final def inputSchemas: mutable.Map[Int, Schema] =
+    OperatorExecutor.inputSchemas.computeIfAbsent(this, _ => mutable.Map())
 }
 
 object OperatorExecutor {
@@ -100,4 +113,6 @@ object OperatorExecutor {
   private val states = new MapMaker().weakKeys().makeMap[OperatorExecutor, State]()
   private[executor] val bindings =
     new MapMaker().weakKeys().makeMap[OperatorExecutor, StateReferenceBinding]()
+  private val inputSchemas =
+    new MapMaker().weakKeys().makeMap[OperatorExecutor, mutable.Map[Int, Schema]]()
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SamplingHelpers.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SamplingHelpers.scala
index 1fa9210898b..6c81befdffd 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SamplingHelpers.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SamplingHelpers.scala
@@ -25,18 +25,12 @@ package org.apache.texera.amber.operator
 object SamplingHelpers {
 
   /**
-    * A Python transcription of `java.util.Random`, for operators whose executor
-    * draws from one.
-    *
-    * A sampler decides per row whether to keep it, so which rows survive is
-    * fixed by the exact sequence the generator produces. Seeding Python's
-    * `random` or numpy's with the engine's seed selects a different set, and
-    * the script would then report a different sample than the workflow it came
-    * from. Only the same generator gives the same rows.
+    * `_TexeraJavaRandom`: `java.util.Random` in Python, so a sampler keeps the rows the engine
+    * keeps. Python's or numpy's generator with the same seed picks other rows. `next_int`
+    * rejects a draw where Java's int sum overflows, which Python's integers never do.
     */
   val JavaRandom: String =
-    """# java.util.Random, transcribed so sampling matches the engine.
-      |class _TexeraJavaRandom:
+    """class _TexeraJavaRandom:
       |    _MASK = (1 << 48) - 1
       |    _MULTIPLIER = 0x5DEECE66D
       |    _ADDEND = 0xB
@@ -60,8 +54,6 @@ object SamplingHelpers {
       |        while True:
       |            bits = self._next(31)
       |            value = bits % bound
-      |            # Java rejects a draw by letting this sum overflow an int.
-      |            # Python would carry it and never reject.
       |            probe = bits - value + (bound - 1)
       |            if ((probe + (1 << 31)) % (1 << 32)) - (1 << 31) >= 0:
       |                return value""".stripMargin
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/cartesianProduct/CartesianProductOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
index 16005c67a59..1f6aaf36e24 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/cartesianProduct/CartesianProductOpDesc.scala
@@ -23,10 +23,11 @@ import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{Attribute, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator}
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
-class CartesianProductOpDesc extends LogicalOp {
+class CartesianProductOpDesc extends LogicalOp with StandaloneCodeGenerator {
+
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -103,4 +104,27 @@ class CartesianProductOpDesc extends LogicalOp {
       ),
       outputPorts = List(OutputPort())
     )
+
+  // Schema mirrors SchemaPropagationFunc: left columns kept as-is, each right
+  // column renamed by repeatedly appending "#@1" while the candidate name
+  // collides with any left column OR any other right column's ORIGINAL name.
+  // The renamed-name table is recomputed at runtime from the actual DataFrame
+  // columns. Known divergence: row order — pandas cross-merge varies right
+  // fastest (L1R1, L1R2, L2R1, L2R2); the JVM op buffers left and emits per
+  // arriving right tuple (L1R1, L2R1, L1R2, L2R2). Cartesian product is set-
+  // semantically order-agnostic, so this is acceptable.
+  override def generateStandaloneCode(): String = {
+    """_left_cols = list(in1df.columns)
+      |_right_cols = list(in2df.columns)
+      |_left_set = set(_left_cols)
+      |_right_set = set(_right_cols)
+      |_rename = {}
+      |for _col in _right_cols:
+      |    _new = _col
+      |    _others = _right_set - {_col}
+      |    while _new in _left_set or _new in _others:
+      |        _new = _new + "#@1"
+      |    _rename[_col] = _new
+      |out1df = in1df.merge(in2df.rename(columns=_rename), how="cross").reset_index(drop=True)""".stripMargin
+  }
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/difference/DifferenceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/difference/DifferenceOpDesc.scala
index c12f057a25b..765939bb85f 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/difference/DifferenceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/difference/DifferenceOpDesc.scala
@@ -23,11 +23,10 @@ import com.google.common.base.Preconditions
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 
-class DifferenceOpDesc extends LogicalOp {
-
+class DifferenceOpDesc extends LogicalOp with StandaloneCodeGenerator {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -61,4 +60,13 @@ class DifferenceOpDesc extends LogicalOp {
       ),
       outputPorts = List(OutputPort(blocking = true))
     )
+
+  // Distinct left (port 0) rows the right (port 1) does not hold, by the
+  // engine's row equality, which `_texera_key_codes` with `signed` numbers.
+  override def generateStandaloneCode(): String =
+    """_texera_set_left, _texera_set_right = _texera_key_codes(in1df, in2df, signed=True)
+      |_texera_set_kept = ~_texera_set_left.duplicated() & ~_texera_set_left.isin(_texera_set_right)
+      |out1df = in1df[_texera_set_kept.to_numpy()].reset_index(drop=True)""".stripMargin
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala
index e8ef7c8bbce..f68c691724c 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala
@@ -23,20 +23,21 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
 import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.state.NoLoopVariable
-import org.apache.texera.amber.core.tuple.{Attribute, Schema}
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{
   ExecutionIdentity,
   PhysicalOpIdentity,
   WorkflowIdentity
 }
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME
 import org.apache.texera.amber.operator.metadata.annotations.{
   AutofillAttributeName,
   AutofillAttributeNameOnPort1
 }
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 object HashJoinOpDesc {
@@ -54,8 +55,7 @@ object HashJoinOpDesc {
   }
 }
 """)
-class HashJoinOpDesc[K] extends LogicalOp {
-
+class HashJoinOpDesc[K] extends LogicalOp with StandaloneCodeGenerator {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Left Input Attribute")
   @JsonPropertyDescription("attribute to be joined on the Left Input")
@@ -158,19 +158,19 @@ class HashJoinOpDesc[K] extends LogicalOp {
               buildSchema.getAttributes.filterNot(_.getName == HASH_JOIN_INTERNAL_KEY_NAME)
             val leftAttributeNames = leftAttributes.map(_.getName).toSet
 
-            // Filter and rename attributes from the probe schema to avoid conflicts
+            // Filter and rename attributes from the probe schema to avoid conflicts,
+            // by the rule the executor names its fields with. Checking the left
+            // side alone declared two columns under one name when the right side
+            // already carried the suffixed one, and the row, matched to the schema
+            // by name, lost the value the executor had put under the other name.
+            val rightNames = probeSchema.getAttributeNames.filterNot(_ == probeAttributeName)
             val rightAttributes = probeSchema.getAttributes
               .filterNot(_.getName == probeAttributeName)
               .map { attr =>
-                var newName = attr.getName
-                while (leftAttributeNames.contains(newName)) {
-                  val suffixIndex = """#@(\d+)$""".r
-                    .findFirstMatchIn(newName)
-                    .map(_.group(1).toInt + 1)
-                    .getOrElse(1)
-                  newName = s"${attr.getName}#@$suffixIndex"
-                }
-                new Attribute(newName, attr.getType)
+                new Attribute(
+                  JoinUtils.renamed(attr.getName, leftAttributeNames.toSeq, rightNames),
+                  attr.getType
+                )
               }
 
             // Combine left and right attributes into a new schema
@@ -203,4 +203,131 @@ class HashJoinOpDesc[K] extends LogicalOp {
       ),
       outputPorts = List(OutputPort())
     )
+
+  // Equi-join: the probe key is dropped and a colliding right column takes
+  // "#@1", as JoinUtils names it. Rows come out in the merge's order rather
+  // than the engine's.
+  /** Only the declared type can say which columns an outer join widened, so
+    * without a schema the widening stands.
+    */
+  override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = {
+    val integral = (a: Attribute) =>
+      a.getType == AttributeType.INTEGER || a.getType == AttributeType.LONG
+    val declaredIntegers = (port: PortIdentity) =>
+      inputSchemas.get(port).map(_.getAttributes.filter(integral).map(_.getName)).getOrElse(List())
+    // Each side keeps its own list, under the names that side's frame carries,
+    // so a column the merge renames is still the column its own schema typed.
+    if (joinType == JoinType.INNER) generateStandaloneCode()
+    else
+      generateStandaloneCode(
+        declaredIntegers(operatorInfo.inputPorts.head.id),
+        declaredIntegers(operatorInfo.inputPorts.last.id)
+      )
+  }
+
+  override def generateStandaloneCode(): String =
+    generateStandaloneCode(List(), List())
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
+
+  private def generateStandaloneCode(
+      leftIntegerColumns: List[String],
+      rightIntegerColumns: List[String]
+  ): String = {
+    val buildKeyLit = objectMapper.writeValueAsString(buildAttributeName)
+    val probeKeyLit = objectMapper.writeValueAsString(probeAttributeName)
+    val how = joinType match {
+      case JoinType.INNER       => "inner"
+      case JoinType.LEFT_OUTER  => "left"
+      case JoinType.RIGHT_OUTER => "right"
+      case JoinType.FULL_OUTER  => "outer"
+    }
+    // An unmatched row leaves a hole, and a hole costs a pandas integer column
+    // its type: int64 becomes float64, which rounds every value past 2^53
+    // before anything can put the type back. The engine writes a null and
+    // leaves the column INTEGER, so widen to the integer dtype that holds a
+    // hole before the merge digs one.
+    val widen = leftIntegerColumns.nonEmpty || rightIntegerColumns.nonEmpty
+    val namesLit = (names: List[String]) => names.map(pyStringLiteral).mkString("[", ", ", "]")
+    val widening =
+      if (!widen) ""
+      else
+        s"""_left_ints = {_c: "Int64" for _c in ${namesLit(leftIntegerColumns)} if _c in in1df.columns}
+           |_right_ints = {_c: "Int64" for _c in ${namesLit(rightIntegerColumns)} if _c in in2df.columns}
+           |""".stripMargin
+    val leftFrame = if (widen) "in1df.astype(_left_ints)" else "in1df"
+    // Cast before the rename, because the names above are the ones the right
+    // input declared.
+    val rightFrame =
+      if (widen) "in2df.astype(_right_ints).rename(columns=_rename)"
+      else "in2df.rename(columns=_rename)"
+    // HashJoinProbeOpExec's rename, written out: append "#@1" until the name is
+    // free. pandas' `suffixes` appends once and then refuses the duplicate it
+    // just made.
+    //
+    // JoinUtils sets the probe key aside before it renames anything, so a
+    // payload column never has to step around the key that is about to leave.
+    // The key itself still rides along, and it only has to land where nothing
+    // else already sits.
+    //
+    // The merge joins on codes and not on the keys. The probe looks each key up
+    // in a HashMap under Scala's ==, where a NaN finds nothing, not even another
+    // NaN, while 0.0 meets -0.0 and a null meets a null; merge pairs NaN with
+    // NaN. `_texera_key_codes` from StandaloneHelpers.AttributeCasts gives each
+    // NaN a code of its own.
+    val merge =
+      s"""_left_cols = set(in1df.columns)
+         |_right_cols = list(in2df.columns)
+         |_payload = [_c for _c in _right_cols if _c != $probeKeyLit]
+         |_payload_set = set(_payload)
+         |_rename = {}
+         |for _col in _payload:
+         |    _new = _col
+         |    _others = _payload_set - {_col}
+         |    while _new in _left_cols or _new in _others:
+         |        _new = _new + "#@1"
+         |    _rename[_col] = _new
+         |_probe_key = $probeKeyLit
+         |_taken = _left_cols | set(_rename.values())
+         |while _probe_key in _taken:
+         |    _probe_key = _probe_key + "#@1"
+         |if _probe_key != $probeKeyLit:
+         |    _rename[$probeKeyLit] = _probe_key
+         |_join_key = "_texera_join_key"
+         |while _join_key in _taken or _join_key == _probe_key:
+         |    _join_key = _join_key + "_"
+         |_left_codes, _right_codes = _texera_key_codes(in1df[[$buildKeyLit]], in2df[[$probeKeyLit]])
+         |""".stripMargin + widening +
+        s"""out1df = $leftFrame.assign(**{_join_key: _left_codes}).merge(
+           |    $rightFrame.assign(**{_join_key: _right_codes}),
+           |    how=${pyStringLiteral(how)},
+           |    on=_join_key,
+           |).drop(columns=[_join_key])""".stripMargin
+    // `_probe_key` is whatever the rename settled on, so this drops the right
+    // column rather than a left one that happened to share its name. Asked of
+    // the name the rename produced and not of the two the operator was given:
+    // when both sides name the key alike the rename still moves the right one
+    // aside, and a test on the operator's own two names read that as one shared
+    // column and kept the copy, where the engine emits the key once.
+    //
+    // Where both sides name the key alike, the engine's unmatched right row
+    // fills the left key column with its own key. The merge left that column
+    // empty and put the key in the renamed right one, which is about to go.
+    // A matched row holds the same value in both, so filling the holes changes
+    // only the rows the right side alone brought.
+    val rightRowsKept = joinType == JoinType.RIGHT_OUTER || joinType == JoinType.FULL_OUTER
+    val fillKey =
+      if (rightRowsKept && buildAttributeName == probeAttributeName)
+        s"""out1df[$buildKeyLit] = out1df[$buildKeyLit].where(
+           |    out1df[$buildKeyLit].notna(), out1df[_probe_key]
+           |)
+           |""".stripMargin
+      else ""
+    val tail =
+      fillKey +
+        s"""if _probe_key != $buildKeyLit:
+           |    out1df = out1df.drop(columns=[_probe_key])
+           |out1df = out1df.reset_index(drop=True)""".stripMargin
+    s"$merge\n$tail"
+  }
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala
index a27ae1d5e24..c11bcb19ed4 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala
@@ -28,6 +28,20 @@ import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
 
 object JoinUtils {
+
+  /** The name a right column takes in the joined row: "#@1" appended until it is
+    * free of every left name and of every other right name. The schema the
+    * operator declares, a matched row and an unmatched right row all name their
+    * right columns by this one rule, since the row is matched to the schema by
+    * name and a column the two name apart loses its value.
+    */
+  def renamed(name: String, leftNames: Seq[String], rightNames: Seq[String]): String = {
+    val others = rightNames.filterNot(_ == name)
+    var newName = name
+    while (leftNames.contains(newName) || others.contains(newName)) newName = s"$newName#@1"
+    newName
+  }
+
   def joinTuples(
       leftTuple: Tuple,
       rightTuple: Tuple,
@@ -45,15 +59,7 @@ object JoinUtils {
     // Create a Map from rightTuple's fields, renaming conflicts
     val rightTupleFields = rightAttributeNames
       .map { name =>
-        var newName = name
-        while (
-          leftAttributeNames.contains(newName) || rightAttributeNames
-            .filter(attrName => name != attrName)
-            .contains(newName)
-        ) {
-          newName = s"$newName#@1"
-        }
-        newName -> rightTuple.getField[Any](name)
+        renamed(name, leftAttributeNames, rightAttributeNames) -> rightTuple.getField[Any](name)
       }
 
     TupleLike(leftTupleFields ++ rightTupleFields)
@@ -68,6 +74,16 @@ class HashJoinProbeOpExec[K](
     objectMapper.readValue(descString, classOf[HashJoinOpDesc[K]])
   var buildTableHashMap: mutable.HashMap[K, (ListBuffer[Tuple], Boolean)] = _
 
+  // The build side's column names, from the schema its port declared. An
+  // unmatched right row has no build row of its own to name them, and a worker
+  // can receive no build rows at all. The build port finishes before the first
+  // right row arrives.
+  private def leftAttributeNames: List[String] =
+    inputSchemas
+      .get(0)
+      .map(_.getAttributeNames.filterNot(_ == HASH_JOIN_INTERNAL_KEY_NAME))
+      .getOrElse(List.empty)
+
   override def open(): Unit = {
     buildTableHashMap = new mutable.HashMap[K, (mutable.ListBuffer[Tuple], Boolean)]()
   }
@@ -142,11 +158,24 @@ class HashJoinProbeOpExec[K](
     }
   }
 
-  private def performRightAntiJoin(tuple: Tuple): Iterator[TupleLike] =
+  // The right columns are named as a matched row names them, so a column the
+  // left side also names goes under its suffixed name and not into the left
+  // column. The probe key fills the left key column where both sides name the
+  // key alike, and is left out where a left payload shares its name, which it
+  // would otherwise fill.
+  private def performRightAntiJoin(tuple: Tuple): Iterator[TupleLike] = {
+    val keyName = desc.probeAttributeName
+    val rightNames = tuple.getSchema.getAttributeNames.filterNot(_ == keyName)
+    val key =
+      if (keyName == desc.buildAttributeName || !leftAttributeNames.contains(keyName))
+        Seq(keyName -> tuple.getField[Any](keyName))
+      else Seq.empty
     Iterator(
       TupleLike(
-        tuple.getSchema.getAttributeNames
-          .map(attributeName => attributeName -> tuple.getField(attributeName)): _*
+        key ++ rightNames.map { name =>
+          JoinUtils.renamed(name, leftAttributeNames, rightNames) -> tuple.getField[Any](name)
+        }: _*
       )
     )
+  }
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intersect/IntersectOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intersect/IntersectOpDesc.scala
index 323d51b0ad5..1a68b0f8cfc 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intersect/IntersectOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intersect/IntersectOpDesc.scala
@@ -22,11 +22,10 @@ package org.apache.texera.amber.operator.intersect
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 
-class IntersectOpDesc extends LogicalOp {
-
+class IntersectOpDesc extends LogicalOp with StandaloneCodeGenerator {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -52,4 +51,13 @@ class IntersectOpDesc extends LogicalOp {
       inputPorts = List(InputPort(PortIdentity()), InputPort(PortIdentity(1))),
       outputPorts = List(OutputPort(blocking = true))
     )
+
+  // Distinct left rows the right also holds, by the engine's row equality, which
+  // `_texera_key_codes` with `signed` numbers.
+  override def generateStandaloneCode(): String =
+    """_texera_set_left, _texera_set_right = _texera_key_codes(in1df, in2df, signed=True)
+      |_texera_set_kept = ~_texera_set_left.duplicated() & _texera_set_left.isin(_texera_set_right)
+      |out1df = in1df[_texera_set_kept.to_numpy()].reset_index(drop=True)""".stripMargin
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
 }
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
index 64c0ac435b3..9f61db9d70b 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala
@@ -24,10 +24,10 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle}
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.state.NoLoopVariable
-import org.apache.texera.amber.core.tuple.{Attribute, Schema}
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 import org.apache.texera.amber.operator.metadata.annotations.{
   AutofillAttributeName,
   AutofillAttributeNameOnPort1
@@ -53,8 +53,7 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper
   }
 }
 """)
-class IntervalJoinOpDesc extends LogicalOp {
-
+class IntervalJoinOpDesc extends LogicalOp with StandaloneCodeGenerator {
   @JsonProperty(required = true)
   @JsonSchemaTitle("Left Input attr")
   @JsonPropertyDescription("Choose one attribute in the left table")
@@ -150,6 +149,63 @@ class IntervalJoinOpDesc extends LogicalOp {
       outputPorts = List(OutputPort())
     )
 
+  // Inner interval join: left point in [rightKey, rightKey + constant], bounds
+  // toggled by include{Left,Right}Bound. Cross-join + mask computes the full
+  // result (no sorted-input assumption, unlike the exec). Runtime dtype check
+  // picks numeric vs pd.DateOffset (unit from timeIntervalType).
+  //
+  // Both keys are read off the merged frame rather than copied into columns of
+  // the operator's own naming, which would overwrite an input column that
+  // happens to carry that name and then drop it. Only the right key needs
+  // locating: the merge leaves every left name alone and suffixes a right one
+  // that collides.
+  override def generateStandaloneCode(): String = generateStandaloneCode(Map.empty)
+
+  override def generateStandaloneCode(inputSchemas: Map[PortIdentity, Schema]): String = {
+    val leftLit = objectMapper.writeValueAsString(leftAttributeName)
+    val rightLit = objectMapper.writeValueAsString(rightAttributeName)
+    val loOp = if (includeLeftBound) ">=" else ">"
+    val hiOp = if (includeRightBound) "<=" else "<"
+    val offsetUnit = Option(timeIntervalType).flatten match {
+      case Some(TimeIntervalType.YEAR)   => "years"
+      case Some(TimeIntervalType.MONTH)  => "months"
+      case Some(TimeIntervalType.HOUR)   => "hours"
+      case Some(TimeIntervalType.MINUTE) => "minutes"
+      case Some(TimeIntervalType.SECOND) => "seconds"
+      case _                             => "days" // DAY or unset
+    }
+    // The engine adds the constant in the right key's declared type. An INTEGER
+    // bound is an Int sum, so the constant is cut to 32 bits and the sum wraps
+    // past Int.MaxValue, where pandas holds the column as int64 and does
+    // neither. A LONG wraps in int64 as it does in Java. The key is widened to
+    // float64 first, which holds every such sum exactly, since NumPy refuses
+    // 2**31 against an int32 column.
+    val declared = inputSchemas
+      .get(operatorInfo.inputPorts.last.id)
+      .flatMap(schema => scala.util.Try(schema.getAttribute(rightAttributeName)).toOption)
+      .map(_.getType)
+    val numericBound =
+      if (declared.contains(AttributeType.INTEGER))
+        s"""(_iv_r.astype("float64") + ${constant.toInt} + 2**31) % 2**32 - 2**31"""
+      else s"_iv_r + $constant"
+    // Timestamp keys arrive as wall clocks, and the engine adds the constant to
+    // the wall clock. It compares by getTime, so a moment counts only to the
+    // millisecond. The bound is added before it is cut, as there.
+    s"""_pairs = in1df.merge(in2df, how="cross", suffixes=("", "#@1"))
+       |_iv_l = _pairs[$leftLit]
+       |_iv_r = _pairs[$rightLit if $rightLit not in in1df.columns else $rightLit + "#@1"]
+       |if pd.api.types.is_datetime64_any_dtype(_iv_r):
+       |    _iv_hi = _texera_wall_clock_instants(_iv_r + pd.DateOffset($offsetUnit=$constant))
+       |    _iv_l = _texera_wall_clock_instants(_iv_l)
+       |    _iv_r = _texera_wall_clock_instants(_iv_r)
+       |else:
+       |    _iv_hi = $numericBound
+       |_iv_match = (_iv_l $loOp _iv_r) & (_iv_l $hiOp _iv_hi)
+       |out1df = _pairs[_iv_match].reset_index(drop=True)""".stripMargin
+  }
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
+
   def this(
       leftTableAttributeName: String,
       rightTableAttributeName: String,
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
index d0d6deecbe6..e438ce6c608 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDesc.scala
@@ -23,11 +23,10 @@ import com.google.common.base.Preconditions
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow._
-import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers}
 
-class SymmetricDifferenceOpDesc extends LogicalOp {
-
+class SymmetricDifferenceOpDesc extends LogicalOp with StandaloneCodeGenerator {
   override def getPhysicalOp(
       workflowId: WorkflowIdentity,
       executionId: ExecutionIdentity
@@ -62,4 +61,16 @@ class SymmetricDifferenceOpDesc extends LogicalOp {
       outputPorts = List(OutputPort(blocking = true))
     )
 
+  // Distinct rows in exactly one input, by the engine's row equality, which
+  // `_texera_key_codes` with `signed` numbers.
+  override def generateStandaloneCode(): String =
+    """_texera_set_left, _texera_set_right = _texera_key_codes(in1df, in2df, signed=True)
+      |_texera_set_left_kept = ~_texera_set_left.duplicated() & ~_texera_set_left.isin(_texera_set_right)
+      |_texera_set_right_kept = ~_texera_set_right.duplicated() & ~_texera_set_right.isin(_texera_set_left)
+      |out1df = pd.concat(
+      |    [in1df[_texera_set_left_kept.to_numpy()], in2df[_texera_set_right_kept.to_numpy()]],
+      |    ignore_index=True,
+      |)""".stripMargin
+
+  override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.AttributeCasts)
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/difference/DifferenceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/difference/DifferenceOpDescSpec.scala
index f6d17e94ebc..11f7c875421 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/difference/DifferenceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/difference/DifferenceOpDescSpec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.texera.amber.operator.difference
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
@@ -29,11 +30,22 @@ import org.apache.texera.amber.core.workflow.{
   UnknownPartition
 }
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.tags.IntegrationTest
+import org.apache.texera.amber.testsupport.setop.SetRows
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class DifferenceOpDescSpec extends AnyFlatSpec with Matchers {
 
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
+
   private val workflowId = WorkflowIdentity(1L)
   private val executionId = ExecutionIdentity(1L)
 
@@ -140,4 +152,88 @@ class DifferenceOpDescSpec extends AnyFlatSpec with Matchers {
     val b = new DifferenceOpDesc
     a.operatorIdentifier should not equal b.operatorIdentifier
   }
+
+  // A missing value and a stored NaN are two different values to the engine, so
+  // a left holding both against a right holding only the missing one keeps the
+  // NaN row. The two are only distinct in a nullable dtype, which is what an
+  // Arrow file is read into.
+  "DifferenceOpDesc.generateStandaloneCode" should
+    "keep the NaN row that a null row does not cancel" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val driver =
+      s"""import pandas as pd, numpy as np
+         |
+         |def col(vals, mask):
+         |    return pd.arrays.FloatingArray(np.array(vals), np.array(mask))
+         |
+         |in1df = pd.DataFrame({"v": col([0.0, np.nan], [True, False])})
+         |in2df = pd.DataFrame({"v": col([0.0], [True])})
+         |${exported()}
+         |print(len(out1df), [("NULL" if x is pd.NA else repr(float(x))) for x in out1df["v"]])
+         |""".stripMargin
+
+    val script = Files.createTempFile("difference-null-nan-", ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+
+    withClue(s"python said:\n$out\nscript:\n$driver") {
+      process.exitValue() shouldBe 0
+      out.trim shouldBe "1 ['nan']"
+    }
+  }
+
+  // The engine's HashSet compares rows with Scala's ==, under which NaN equals
+  // nothing, so both left NaN rows survive, and hashes them by each double's
+  // bits, so a left 0.0 survives a right -0.0. A finite value and a null on
+  // both sides are the controls.
+  it should "match rows by the engine's equality for NaN and signed zero" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val exec = new DifferenceOpExec
+    exec.open()
+    SetRows.left.foreach(t => exec.processTuple(t, 0))
+    SetRows.right.foreach(t => exec.processTuple(t, 1))
+    exec.onFinish(0)
+    val engine = SetRows.rendered(exec.onFinish(1))
+
+    val out = SetRows.runExport(python, exported(), "difference-nan-zero-")
+    withClue(s"script:\n${exported()}")(out shouldBe engine)
+  }
+
+  private def exported(): String = {
+    val desc = new DifferenceOpDesc
+    (desc.standaloneHelpers() :+ desc.generateStandaloneCode()).mkString("\n")
+  }
+
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start()
+    ).toOption.exists { p =>
+      if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+      else p.exitValue() == 0
+    }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDescSpec.scala
index 4ead90b6456..ae1ceae57db 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDescSpec.scala
@@ -19,16 +19,33 @@
 
 package org.apache.texera.amber.operator.hashJoin
 
-import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import com.typesafe.config.ConfigFactory
+import org.apache.texera.amber.core.tuple.{
+  Attribute,
+  AttributeType,
+  Schema,
+  SchemaEnforceable,
+  Tuple
+}
 import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.tags.IntegrationTest
 import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class HashJoinOpDescSpec extends AnyFlatSpec with Matchers {
 
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
+
   private def leftRight(): (Schema, Schema) =
     (
       Schema()
@@ -75,6 +92,146 @@ class HashJoinOpDescSpec extends AnyFlatSpec with Matchers {
     out(d.operatorInfo.outputPorts.head.id).getAttributeNames shouldBe List("a", "k", "k#@1")
   }
 
+  // A key that is missing and a key holding a NaN are two different keys to the
+  // engine, so only the missing one matches a missing right key. The two are
+  // only distinct in a nullable dtype, which is what an Arrow file is read into.
+  "HashJoinOpDesc.generateStandaloneCode" should "match a missing key but not a NaN one" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val d = new HashJoinOpDesc[Integer]
+    d.buildAttributeName = "k"
+    d.probeAttributeName = "k"
+    d.joinType = JoinType.LEFT_OUTER
+
+    val driver =
+      s"""import pandas as pd, numpy as np
+         |
+         |def col(vals, mask):
+         |    return pd.arrays.FloatingArray(np.array(vals), np.array(mask))
+         |
+         |in1df = pd.DataFrame({"k": col([0.0, np.nan], [True, False]), "lv": [1, 2]})
+         |in2df = pd.DataFrame({"k": col([0.0], [True]), "rv": [9]})
+         |${d.standaloneHelpers().mkString("\n")}
+         |${d.generateStandaloneCode()}
+         |print(len(out1df), int(out1df["rv"].notna().sum()))
+         |""".stripMargin
+
+    val script = Files.createTempFile("hashjoin-null-nan-", ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    withClue(s"python said:\n$out\nscript:\n$driver") {
+      process.exitValue() shouldBe 0
+      // Both left rows come back, and exactly one of them found a partner.
+      out.trim shouldBe "2 1"
+    }
+  }
+
+  // The probe looks each key up in a HashMap under Scala's ==: a NaN finds
+  // nothing, not even another NaN, while 0.0 meets -0.0 and a null meets a
+  // null. Each join type keeps or drops the unmatched NaN rows by its own rule.
+  for (joinType <- JoinType.values) {
+    it should s"pair NaN, signed-zero and null keys as the engine does in a $joinType join" taggedAs NeedsPythonPackages in {
+      val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+      if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+      val d = new HashJoinOpDesc[java.lang.Double]
+      d.buildAttributeName = "k"
+      d.probeAttributeName = "k2"
+      d.joinType = joinType
+      val left = Schema().add("k", AttributeType.DOUBLE).add("lv", AttributeType.STRING)
+      val right = Schema().add("k2", AttributeType.DOUBLE).add("rv", AttributeType.STRING)
+      val leftRows: Seq[(java.lang.Double, String)] =
+        Seq((Double.NaN, "L-nan"), (0.0, "L+0"), (1.0, "L1"), (null, "L-null"))
+      val rightRows: Seq[(java.lang.Double, String)] =
+        Seq((Double.NaN, "R-nan"), (-0.0, "R-0"), (1.0, "R1"), (null, "R-null"))
+
+      val outSchema = d
+        .getExternalOutputSchemas(Map(PortIdentity() -> left, PortIdentity(1) -> right))
+        .values
+        .head
+      val build = new HashJoinBuildOpExec[java.lang.Double](objectMapper.writeValueAsString(d))
+      build.open()
+      leftRows.foreach { case (k, v) => build.processTuple(Tuple(left, Array[Any](k, v)), 0) }
+      val probe = new HashJoinProbeOpExec[java.lang.Double](objectMapper.writeValueAsString(d))
+      probe.open()
+      val internal =
+        Schema().add(HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME, AttributeType.ANY).add(left)
+      build
+        .onFinish(0)
+        .foreach(t =>
+          probe.processTuple(t.asInstanceOf[SchemaEnforceable].enforceSchema(internal), 0)
+        )
+      probe.onFinish(0)
+      val probed = rightRows.flatMap {
+        case (k, v) => probe.processTuple(Tuple(right, Array[Any](k, v)), 1)
+      } ++ probe.onFinish(1)
+      val engine = probed
+        .map(_.asInstanceOf[SchemaEnforceable].enforceSchema(outSchema).getFields.map {
+          case null                           => "None"
+          case d: java.lang.Double if d.isNaN => "nan"
+          case other                          => other.toString
+        })
+        .map(_.mkString(" "))
+        .toList
+        .sorted
+
+      def frame(rows: Seq[(java.lang.Double, String)], key: String, value: String): String = {
+        val keys =
+          rows.map(_._1).map(k => if (k == null) "None" else if (k.isNaN) "np.nan" else k.toString)
+        val values = rows.map(r => s""""${r._2}"""")
+        s"""pd.DataFrame({
+           |    "$key": col([${keys.mkString(", ")}]),
+           |    "$value": [${values.mkString(", ")}],
+           |})""".stripMargin
+      }
+      val driver =
+        s"""import pandas as pd, numpy as np
+           |
+           |def col(vals):
+           |    return pd.arrays.FloatingArray(
+           |        np.array([0.0 if v is None else v for v in vals]),
+           |        np.array([v is None for v in vals]),
+           |    )
+           |
+           |in1df = ${frame(leftRows, "k", "lv")}
+           |in2df = ${frame(rightRows, "k2", "rv")}
+           |${d.standaloneHelpers().mkString("\n")}
+           |${d.generateStandaloneCode(Map(PortIdentity() -> left, PortIdentity(1) -> right))}
+           |def show(v):
+           |    if isinstance(v, float):
+           |        return "nan" if v != v else repr(float(v))
+           |    return str(v)
+           |rows = []
+           |for values, missing in zip(out1df.itertuples(index=False), out1df.isna().to_numpy()):
+           |    rows.append(" ".join("None" if m else show(v) for v, m in zip(values, missing)))
+           |print("\\n".join(sorted(rows)))
+           |""".stripMargin
+
+      val script = Files.createTempFile("hashjoin-nan-zero-", ".py")
+      script.toFile.deleteOnExit()
+      Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+      val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+      val out = Source.fromInputStream(process.getInputStream).mkString
+      process.waitFor(120, TimeUnit.SECONDS)
+      withClue(s"python said:\n$out\nscript:\n$driver") {
+        process.exitValue() shouldBe 0
+        out.trim.linesIterator.toList shouldBe engine
+      }
+    }
+  }
+
+  // Without a schema nothing says which columns were integers.
+  it should "leave the widening alone when no schema is given" in {
+    val d = new HashJoinOpDesc[Integer]
+    d.buildAttributeName = "k"
+    d.probeAttributeName = "kk"
+    d.joinType = JoinType.FULL_OUTER
+    d.generateStandaloneCode() should not include "astype("
+  }
   "HashJoinOpDesc" should "round-trip its config fields through the polymorphic base" in {
     val d = new HashJoinOpDesc[String]
     d.buildAttributeName = "lk"
@@ -89,4 +246,33 @@ class HashJoinOpDescSpec extends AnyFlatSpec with Matchers {
     r.probeAttributeName shouldBe "rk"
     r.joinType shouldBe JoinType.LEFT_OUTER
   }
+
+  // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path
+  // (UDF_PYTHON_PATH), then python3 / python / py.
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start()
+    ).toOption
+      .exists { p =>
+        if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+        else p.exitValue() == 0
+      }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExecSpec.scala
index 2a98fd214e9..f249b50c228 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExecSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExecSpec.scala
@@ -20,6 +20,8 @@
 package org.apache.texera.amber.operator.hashJoin
 
 import org.apache.texera.amber.core.tuple._
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.scalatest.flatspec.AnyFlatSpec
@@ -87,6 +89,8 @@ class HashJoinProbeOpExecSpec extends AnyFlatSpec {
       )
     buildExec.close()
 
+    // The runtime records the port's declared schema before finishing it.
+    probeExec.inputSchemas(buildPort) = internalHashTableSchema
     assert(probeExec.onFinish(buildPort).isEmpty)
     probeExec
   }
@@ -274,4 +278,134 @@ class HashJoinProbeOpExecSpec extends AnyFlatSpec {
     probeExec.close()
     assert(probeExec.buildTableHashMap.isEmpty)
   }
+
+  /** The rows a join over `left` and `right` emits, matched to the schema the
+    * operator declares by name, as the engine matches them.
+    */
+  private def joinedRows(
+      joinType: JoinType,
+      left: Schema,
+      leftRows: Seq[Array[Any]],
+      right: Schema,
+      rightRows: Seq[Array[Any]],
+      buildKey: String = "k",
+      probeKey: String = "k"
+  ): (List[String], List[List[Any]]) = {
+    val desc = new HashJoinOpDesc[String]()
+    desc.buildAttributeName = buildKey
+    desc.probeAttributeName = probeKey
+    desc.joinType = joinType
+    val plan = desc.getPhysicalPlan(WorkflowIdentity(1L), ExecutionIdentity(1L))
+    val build = plan.operators.find(_.id.layerName == "build").get
+    val probe = plan.operators.find(_.id.layerName == "probe").get
+    val buildOut =
+      build.propagateSchema.func(Map(PortIdentity() -> left))(PortIdentity(0, internal = true))
+    val declared = probe.propagateSchema
+      .func(Map(PortIdentity(0, internal = true) -> buildOut, PortIdentity(1) -> right))(
+        PortIdentity()
+      )
+
+    val descString = objectMapper.writeValueAsString(desc)
+    val buildExec = new HashJoinBuildOpExec[String](descString)
+    buildExec.open()
+    leftRows.foreach(r => buildExec.processTuple(Tuple.builder(left).addSequentially(r).build(), 0))
+    val probeExec = new HashJoinProbeOpExec[String](descString)
+    probeExec.open()
+    buildExec
+      .onFinish(0)
+      .foreach(t =>
+        probeExec.processTuple(t.asInstanceOf[SchemaEnforceable].enforceSchema(buildOut), 0)
+      )
+    // The runtime records the port's declared schema before finishing it.
+    probeExec.inputSchemas(0) = buildOut
+    probeExec.onFinish(0)
+    val emitted = rightRows.flatMap(r =>
+      probeExec.processTuple(Tuple.builder(right).addSequentially(r).build(), 1)
+    ) ++ probeExec.onFinish(1)
+    (
+      declared.getAttributeNames,
+      emitted.map(_.asInstanceOf[SchemaEnforceable].enforceSchema(declared).getFields.toList).toList
+    )
+  }
+
+  // A right column meeting a left name takes "#@1", and the right side can
+  // already hold that name, as the output of an earlier join does. The schema
+  // declared two columns under it while the executor named the row "#@1#@1",
+  // so matching the row to the schema lost the right x and doubled the other.
+  it should "name a right column colliding twice alike in the schema and the row" in {
+    val (names, rows) = joinedRows(
+      JoinType.INNER,
+      Schema().add("k", AttributeType.STRING).add("x", AttributeType.STRING),
+      Seq(Array[Any]("1", "left")),
+      Schema()
+        .add("k", AttributeType.STRING)
+        .add("x", AttributeType.STRING)
+        .add("x#@1", AttributeType.STRING),
+      Seq(Array[Any]("1", "right x", "right x#@1"))
+    )
+    assert(names == List("k", "x", "x#@1#@1", "x#@1"))
+    assert(rows == List(List("1", "left", "right x", "right x#@1")))
+  }
+
+  // An unmatched right row kept its columns' own names, so one the left side
+  // also names landed in the left column, and failed there on a type the left
+  // one did not declare. Its key still fills the left key column.
+  it should "name an unmatched right row's columns as a joined row names them" in {
+    val (names, rows) = joinedRows(
+      JoinType.FULL_OUTER,
+      Schema().add("k", AttributeType.STRING).add("x", AttributeType.INTEGER),
+      Seq(Array[Any]("1", 10)),
+      Schema().add("k", AttributeType.STRING).add("x", AttributeType.DOUBLE),
+      Seq(Array[Any]("2", 2.5))
+    )
+    assert(names == List("k", "x", "x#@1"))
+    assert(rows.toSet == Set(List("2", null, 2.5), List("1", 10, null)))
+  }
+
+  // The key an unmatched right row carries fills the left key column only when
+  // the two keys share a name. Under any other name it belonged to no declared
+  // column, and where a left payload happened to carry that name the key landed
+  // in the payload.
+  it should "not let an unmatched right key fill a left payload that shares its name" in {
+    val (names, rows) = joinedRows(
+      JoinType.FULL_OUTER,
+      Schema().add("id", AttributeType.STRING).add("key", AttributeType.STRING),
+      Seq(Array[Any]("1", "payload")),
+      Schema().add("key", AttributeType.STRING).add("value", AttributeType.INTEGER),
+      Seq(Array[Any]("2", 9)),
+      buildKey = "id",
+      probeKey = "key"
+    )
+    assert(names == List("id", "key", "value"))
+    assert(rows.toSet == Set(List(null, null, 9), List("1", "payload", null)))
+  }
+
+  // A worker can receive no build rows at all, when the left input is empty or
+  // no left key hashes to it. The left names then come from the build port's
+  // declared schema alone, and the two cases above must still hold.
+  it should "name an unmatched right row's columns when the build side sent no rows" in {
+    val (names, rows) = joinedRows(
+      JoinType.FULL_OUTER,
+      Schema().add("k", AttributeType.STRING).add("x", AttributeType.INTEGER),
+      Seq.empty,
+      Schema().add("k", AttributeType.STRING).add("x", AttributeType.DOUBLE),
+      Seq(Array[Any]("2", 2.5))
+    )
+    assert(names == List("k", "x", "x#@1"))
+    assert(rows == List(List("2", null, 2.5)))
+  }
+
+  it should "not let an unmatched right key fill a left payload when the build side sent no rows" in {
+    val (names, rows) = joinedRows(
+      JoinType.FULL_OUTER,
+      Schema().add("id", AttributeType.STRING).add("key", AttributeType.STRING),
+      Seq.empty,
+      Schema().add("key", AttributeType.STRING).add("value", AttributeType.INTEGER),
+      Seq(Array[Any]("2", 9)),
+      buildKey = "id",
+      probeKey = "key"
+    )
+    assert(names == List("id", "key", "value"))
+    assert(rows == List(List(null, null, 9)))
+  }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intersect/IntersectOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intersect/IntersectOpDescSpec.scala
index d9be82e0cdb..05a4ffa0910 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intersect/IntersectOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intersect/IntersectOpDescSpec.scala
@@ -19,15 +19,27 @@
 
 package org.apache.texera.amber.operator.intersect
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{HashPartition, SinglePartition, UnknownPartition}
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.tags.IntegrationTest
+import org.apache.texera.amber.testsupport.setop.SetRows
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class IntersectOpDescSpec extends AnyFlatSpec with Matchers {
 
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
+
   private val workflowId = WorkflowIdentity(1L)
   private val executionId = ExecutionIdentity(1L)
 
@@ -79,4 +91,87 @@ class IntersectOpDescSpec extends AnyFlatSpec with Matchers {
         fail(s"expected OpExecWithClassName, got $other")
     }
   }
+
+  // A missing value and a stored NaN are two values to the engine's HashSet, so
+  // a left holding both against a right holding only the missing one keeps the
+  // missing row alone. They are only distinct in a nullable dtype, which is what
+  // an Arrow file is read into.
+  "IntersectOpDesc.generateStandaloneCode" should
+    "keep the null row a null row matches and not the NaN row" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val driver =
+      s"""import pandas as pd, numpy as np
+         |
+         |def col(vals, mask):
+         |    return pd.arrays.FloatingArray(np.array(vals), np.array(mask))
+         |
+         |in1df = pd.DataFrame({"v": col([0.0, np.nan], [True, False])})
+         |in2df = pd.DataFrame({"v": col([0.0], [True])})
+         |${exported()}
+         |print(len(out1df), [("NULL" if x is pd.NA else repr(float(x))) for x in out1df["v"]])
+         |""".stripMargin
+
+    val script = Files.createTempFile("intersect-null-nan-", ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+
+    withClue(s"python said:\n$out\nscript:\n$driver") {
+      process.exitValue() shouldBe 0
+      out.trim shouldBe "1 ['NULL']"
+    }
+  }
+
+  // The engine's HashSet compares rows with Scala's ==, under which NaN equals
+  // nothing, and hashes them by each double's bits, which tell 0.0 from -0.0.
+  // A finite value and a null on both sides are the controls.
+  it should "match rows by the engine's equality for NaN and signed zero" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val exec = new IntersectOpExec
+    exec.open()
+    SetRows.left.foreach(t => exec.processTuple(t, 0))
+    SetRows.right.foreach(t => exec.processTuple(t, 1))
+    exec.onFinish(0)
+    val engine = SetRows.rendered(exec.onFinish(1))
+
+    val out = SetRows.runExport(python, exported(), "intersect-nan-zero-")
+    withClue(s"script:\n${exported()}")(out shouldBe engine)
+  }
+
+  private def exported(): String = {
+    val desc = new IntersectOpDesc
+    (desc.standaloneHelpers() :+ desc.generateStandaloneCode()).mkString("\n")
+  }
+
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start()
+    ).toOption.exists { p =>
+      if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+      else p.exitValue() == 0
+    }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDescSpec.scala
index 01d9d2e10b5..aac64c1ba41 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDescSpec.scala
@@ -19,18 +19,30 @@
 
 package org.apache.texera.amber.operator.intervalJoin
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
-import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{HashPartition, PortIdentity}
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.tags.IntegrationTest
 import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.sql.Timestamp
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class IntervalJoinOpDescSpec extends AnyFlatSpec with Matchers {
 
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
+
   private val workflowId = WorkflowIdentity(1L)
   private val executionId = ExecutionIdentity(1L)
 
@@ -104,6 +116,279 @@ class IntervalJoinOpDescSpec extends AnyFlatSpec with Matchers {
       List("a", "k", "b", "k#@1")
   }
 
+  "IntervalJoinOpDesc.generateStandaloneCode" should
+    "read the right key under the name the merge left it with" in {
+    // "rk" is only suffixed when the left frame has a column of that name, and
+    // the frame is not known until the script runs, so the choice is in Python.
+    configured().generateStandaloneCode() should include(
+      """_iv_r = _pairs["rk" if "rk" not in in1df.columns else "rk" + "#@1"]"""
+    )
+  }
+
+  /** The bug this pins: the keys used to be copied into `_iv_l` and `_iv_r`
+    * columns, so an input column already carrying one of those names was
+    * overwritten and then dropped. Run the generated pandas and hold its
+    * columns against the schema the operator promises for the same two inputs.
+    */
+  it should "keep an input column that carries a temporary's name" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val left = Schema()
+      .add(new Attribute("x", AttributeType.INTEGER))
+      .add(new Attribute("_iv_l", AttributeType.STRING))
+    val right = Schema().add(new Attribute("r", AttributeType.INTEGER))
+
+    val d = new IntervalJoinOpDesc
+    d.leftAttributeName = "x"
+    d.rightAttributeName = "r"
+    d.constant = 3L
+    d.timeIntervalType = None
+    val promised = d
+      .getPhysicalOp(workflowId, executionId)
+      .propagateSchema
+      .func(Map(PortIdentity() -> left, PortIdentity(1) -> right))(
+        d.operatorInfo.outputPorts.head.id
+      )
+      .getAttributeNames
+
+    val driver =
+      s"""import pandas as pd
+         |
+         |in1df = pd.DataFrame({"x": [2], "_iv_l": ["payload"]})
+         |in2df = pd.DataFrame({"r": [1]})
+         |${d.generateStandaloneCode()}
+         |print(",".join(map(str, out1df.columns)))
+         |print(",".join(map(str, out1df.iloc[0].tolist())))
+         |""".stripMargin
+
+    val script = Files.createTempFile("intervaljoin-columns-", ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    withClue(s"python said:\n$out\nscript:\n$driver") {
+      process.exitValue() shouldBe 0
+      val lines = out.trim.linesIterator.toSeq
+      lines.head.split(",").toList shouldBe promised
+      // The payload the temporary used to overwrite, and the row still joins.
+      lines(1) shouldBe "2,payload,1"
+    }
+  }
+
+  // IntervalJoinOpExec compares timestamps by getTime, so only whole
+  // milliseconds count. The right key is 12:00:00.0009 with a three-second
+  // interval: the first left row shares its millisecond and the last shares
+  // the upper bound's while past it, so each joins only at that resolution.
+  private val millisecondLefts =
+    Seq("2024-01-01 12:00:00.0005", "2024-01-01 12:00:02", "2024-01-01 12:00:03.00095")
+  private val millisecondRight = "2024-01-01 12:00:00.0009"
+
+  it should "compare timestamps by the millisecond as the engine does" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val left = Schema().add(new Attribute("l", AttributeType.TIMESTAMP))
+    val right = Schema().add(new Attribute("r", AttributeType.TIMESTAMP))
+
+    def check(includeBounds: Boolean): Unit = {
+      val d =
+        new IntervalJoinOpDesc("l", "r", 3L, includeBounds, includeBounds, TimeIntervalType.SECOND)
+
+      val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(d))
+      exec.open()
+      val engine =
+        try {
+          exec.processTuple(Tuple(right, Array[Any](Timestamp.valueOf(millisecondRight))), 1)
+          millisecondLefts.flatMap { l =>
+            exec
+              .processTuple(Tuple(left, Array[Any](Timestamp.valueOf(l))), 0)
+              .map(_.getFields.toSeq)
+              .map(_.map(_.asInstanceOf[Timestamp].toLocalDateTime.toString).mkString(" "))
+          }
+        } finally exec.close()
+
+      def literal(t: String) = "\"" + t.replace(' ', 'T') + "\""
+      val lefts = millisecondLefts.map(literal).mkString("[", ", ", "]")
+      val driver =
+        s"""import numpy as np
+           |import pandas as pd
+           |
+           |in1df = pd.DataFrame({"l": np.array($lefts, dtype="datetime64[us]")})
+           |in2df = pd.DataFrame({"r": np.array([${literal(millisecondRight)}], dtype="datetime64[us]")})
+           |${d.standaloneHelpers().mkString("\n\n")}
+           |${d.generateStandaloneCode()}
+           |for row in out1df.itertuples(index=False):
+           |    print(" ".join(pd.Timestamp(v).isoformat() for v in row))
+           |""".stripMargin
+
+      val script = Files.createTempFile("intervaljoin-millis-", ".py")
+      script.toFile.deleteOnExit()
+      Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+      val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+      val out = Source.fromInputStream(process.getInputStream).mkString
+      process.waitFor(120, TimeUnit.SECONDS)
+      withClue(s"python said:\n$out\nscript:\n$driver") {
+        process.exitValue() shouldBe 0
+        out.trim.linesIterator.toSeq shouldBe engine
+      }
+      if (includeBounds) engine should have size 3
+    }
+
+    check(includeBounds = true)
+    check(includeBounds = false)
+  }
+
+  // A source hands the script each timestamp as the wall clock the machine's
+  // zone shows, with no zone attached, and the executor adds the constant to
+  // that wall clock. So in Chicago a day from 03-09 12:00 ends at 03-10 12:00
+  // across the spring change, and the same holds across the autumn one. A day
+  // into the spring gap ends where Timestamp.valueOf reads 02:30, an hour
+  // later, and a month from 01-31 ends on 02-29. Each case puts one left row
+  // on or before the engine's bound and one past it. Both sides are fed the
+  // same wall clocks, and the script runs with TZ set to the same zone.
+  it should "add a time interval on the machine's wall clock as the engine does" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val zone = "America/Chicago"
+    val cases = Seq(
+      (TimeIntervalType.DAY, "2024-06-01T12:00", Seq("2024-06-02T12:00", "2024-06-02T12:01")),
+      (TimeIntervalType.DAY, "2024-03-09T12:00", Seq("2024-03-10T11:30", "2024-03-10T12:30")),
+      (TimeIntervalType.DAY, "2024-11-02T12:00", Seq("2024-11-03T11:30", "2024-11-03T12:30")),
+      (TimeIntervalType.DAY, "2024-03-09T02:30", Seq("2024-03-10T03:15", "2024-03-10T03:45")),
+      (TimeIntervalType.DAY, "2024-11-02T01:30", Seq("2024-11-03T01:15", "2024-11-03T01:45")),
+      (TimeIntervalType.MONTH, "2024-01-31T00:00", Seq("2024-02-29T00:00", "2024-02-29T00:01")),
+      (TimeIntervalType.DAY, "1850-06-01T12:00", Seq("1850-06-02T12:00", "1850-06-02T12:01"))
+    )
+    val left = Schema().add(new Attribute("l", AttributeType.TIMESTAMP))
+    val right = Schema().add(new Attribute("r", AttributeType.TIMESTAMP))
+    def at(wall: String) = Timestamp.valueOf(java.time.LocalDateTime.parse(wall))
+    def wall(t: Timestamp) =
+      t.toLocalDateTime.format(java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+
+    val saved = java.util.TimeZone.getDefault
+    java.util.TimeZone.setDefault(java.util.TimeZone.getTimeZone(zone))
+    try {
+      cases.foreach {
+        case (unit, r, lefts) =>
+          val d = new IntervalJoinOpDesc("l", "r", 1L, true, true, unit)
+          val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(d))
+          exec.open()
+          val engine =
+            try {
+              exec.processTuple(Tuple(right, Array[Any](at(r))), 1)
+              lefts.flatMap { l =>
+                exec
+                  .processTuple(Tuple(left, Array[Any](at(l))), 0)
+                  .map(_.getFields.map(v => wall(v.asInstanceOf[Timestamp])).mkString(" "))
+              }
+            } finally exec.close()
+
+          def literal(t: String) = "\"" + t + "\""
+          val driver =
+            s"""import numpy as np
+               |import pandas as pd
+               |
+               |in1df = pd.DataFrame({"l": np.array(${lefts
+              .map(literal)
+              .mkString("[", ", ", "]")}, dtype="datetime64[us]")})
+               |in2df = pd.DataFrame({"r": np.array([${literal(r)}], dtype="datetime64[us]")})
+               |${d.standaloneHelpers().mkString("\n\n")}
+               |${d.generateStandaloneCode()}
+               |for row in out1df.itertuples(index=False):
+               |    print(" ".join(pd.Timestamp(v).isoformat() for v in row))
+               |""".stripMargin
+
+          val script = Files.createTempFile("intervaljoin-zone-", ".py")
+          script.toFile.deleteOnExit()
+          Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+          val builder = new ProcessBuilder(python, script.toString).redirectErrorStream(true)
+          builder.environment().put("TZ", zone)
+          val process = builder.start()
+          val out = Source.fromInputStream(process.getInputStream).mkString
+          process.waitFor(120, TimeUnit.SECONDS)
+          withClue(s"$unit from $r, python said:\n$out\nscript:\n$driver") {
+            process.exitValue() shouldBe 0
+            out.trim.linesIterator.filter(_.nonEmpty).toSeq shouldBe engine
+            engine should have size 1
+          }
+      }
+    } finally java.util.TimeZone.setDefault(saved)
+  }
+
+  // The engine adds the constant in the key's declared type: an INTEGER bound
+  // is an Int sum, with the constant cut to 32 bits, and wraps past
+  // Int.MaxValue, and a LONG one wraps past Long.MaxValue. Each case feeds the
+  // right row first and the left rows in ascending order, as the executor
+  // assumes.
+  it should "add the constant in the key's declared type as the engine does" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(
+      cancel("No runnable python executable (udf.conf python.path, python3, python, py)")
+    )
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val cases = Seq(
+      (AttributeType.INTEGER, 10L, 0L, Seq(3L, 20L)),
+      (AttributeType.INTEGER, 10L, 2147483640L, Seq(2147483645L)),
+      (AttributeType.INTEGER, 4294967306L, 0L, Seq(5L, 20L)),
+      (AttributeType.LONG, 10L, 9223372036854775800L, Seq(9223372036854775805L))
+    )
+    cases.foreach {
+      case (attrType, constant, right, lefts) =>
+        val dtype = if (attrType == AttributeType.INTEGER) "int32" else "int64"
+        def value(v: Long): Any = if (attrType == AttributeType.INTEGER) v.toInt else v
+        val leftSchema = Schema().add(new Attribute("l", attrType))
+        val rightSchema = Schema().add(new Attribute("r", attrType))
+        val d = new IntervalJoinOpDesc("l", "r", constant, true, true, TimeIntervalType.DAY)
+
+        val exec = new IntervalJoinOpExec(objectMapper.writeValueAsString(d))
+        exec.open()
+        val engine =
+          try {
+            exec.processTuple(Tuple(rightSchema, Array[Any](value(right))), 1)
+            lefts.flatMap { l =>
+              exec
+                .processTuple(Tuple(leftSchema, Array[Any](value(l))), 0)
+                .map(_.getFields.mkString(" "))
+            }
+          } finally exec.close()
+
+        val code = d.generateStandaloneCode(
+          Map(PortIdentity() -> leftSchema, PortIdentity(1) -> rightSchema)
+        )
+        val driver =
+          s"""import numpy as np
+             |import pandas as pd
+             |
+             |in1df = pd.DataFrame({"l": np.array(${lefts
+            .mkString("[", ", ", "]")}, dtype="$dtype")})
+             |in2df = pd.DataFrame({"r": np.array([$right], dtype="$dtype")})
+             |$code
+             |for row in out1df.itertuples(index=False):
+             |    print(" ".join(str(int(v)) for v in row))
+             |""".stripMargin
+
+        val script = Files.createTempFile("intervaljoin-bound-", ".py")
+        script.toFile.deleteOnExit()
+        Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+        val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+        val out = Source.fromInputStream(process.getInputStream).mkString
+        process.waitFor(120, TimeUnit.SECONDS)
+        withClue(s"$attrType constant $constant, python said:\n$out\nscript:\n$driver") {
+          process.exitValue() shouldBe 0
+          out.trim.linesIterator.filter(_.nonEmpty).toSeq shouldBe engine
+        }
+    }
+  }
+
   "IntervalJoinOpDesc" should "round-trip its fields through the polymorphic base" in {
     val d = configured()
     d.constant = 42L
@@ -118,4 +403,33 @@ class IntervalJoinOpDescSpec extends AnyFlatSpec with Matchers {
     ij.includeLeftBound shouldBe false
     ij.includeRightBound shouldBe false
   }
+
+  // Python resolution follows FilledAreaPlotOpDescSpec: udf.conf python.path
+  // (UDF_PYTHON_PATH), then python3 / python / py.
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas").redirectErrorStream(true).start()
+    ).toOption
+      .exists { p =>
+        if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+        else p.exitValue() == 0
+      }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDescSpec.scala
index e6e2995877e..01270b92a40 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/symmetricDifference/SymmetricDifferenceOpDescSpec.scala
@@ -19,18 +19,30 @@
 
 package org.apache.texera.amber.operator.symmetricDifference
 
+import com.typesafe.config.ConfigFactory
 import org.apache.texera.amber.core.executor.OpExecWithClassName
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
 import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
 import org.apache.texera.amber.core.workflow.{HashPartition, PortIdentity}
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.tags.IntegrationTest
+import org.apache.texera.amber.testsupport.setop.SetRows
 import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.scalatest.Tag
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+import scala.util.Try
+
 class SymmetricDifferenceOpDescSpec extends AnyFlatSpec with Matchers {
 
+  private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName)
+
   private val workflowId = WorkflowIdentity(1L)
   private val executionId = ExecutionIdentity(1L)
 
@@ -86,4 +98,88 @@ class SymmetricDifferenceOpDescSpec extends AnyFlatSpec with Matchers {
     val json = objectMapper.writeValueAsString(new SymmetricDifferenceOpDesc)
     objectMapper.readValue(json, classOf[LogicalOp]) shouldBe a[SymmetricDifferenceOpDesc]
   }
+
+  // A missing value and a stored NaN are two values to the engine's HashSet, so
+  // a left holding both against a right holding only the missing one leaves the
+  // NaN row alone in exactly one input. They are only distinct in a nullable
+  // dtype, which is what an Arrow file is read into.
+  "SymmetricDifferenceOpDesc.generateStandaloneCode" should
+    "keep the NaN row that a null row does not cancel" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val driver =
+      s"""import pandas as pd, numpy as np
+         |
+         |def col(vals, mask):
+         |    return pd.arrays.FloatingArray(np.array(vals), np.array(mask))
+         |
+         |in1df = pd.DataFrame({"v": col([0.0, np.nan], [True, False])})
+         |in2df = pd.DataFrame({"v": col([0.0], [True])})
+         |${exported()}
+         |print(len(out1df), [("NULL" if x is pd.NA else repr(float(x))) for x in out1df["v"]])
+         |""".stripMargin
+
+    val script = Files.createTempFile("symmetric-difference-null-nan-", ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream).mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+
+    withClue(s"python said:\n$out\nscript:\n$driver") {
+      process.exitValue() shouldBe 0
+      out.trim shouldBe "1 ['nan']"
+    }
+  }
+
+  // The engine's HashSet compares rows with Scala's ==, under which NaN equals
+  // nothing, so every NaN row is in exactly one input, and hashes them by each
+  // double's bits, so 0.0 and -0.0 are each in one. A finite value and a null
+  // on both sides are the controls.
+  it should "match rows by the engine's equality for NaN and signed zero" taggedAs NeedsPythonPackages in {
+    val python = resolvePython().getOrElse(cancel("No runnable python executable"))
+    if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas")
+
+    val exec = new SymmetricDifferenceOpExec
+    exec.open()
+    SetRows.left.foreach(t => exec.processTuple(t, 0))
+    SetRows.right.foreach(t => exec.processTuple(t, 1))
+    exec.onFinish(0)
+    val engine = SetRows.rendered(exec.onFinish(1))
+
+    val out = SetRows.runExport(python, exported(), "symmetric-difference-nan-zero-")
+    withClue(s"script:\n${exported()}")(out shouldBe engine)
+  }
+
+  private def exported(): String = {
+    val desc = new SymmetricDifferenceOpDesc
+    (desc.standaloneHelpers() :+ desc.generateStandaloneCode()).mkString("\n")
+  }
+
+  private def resolvePython(): Option[String] = {
+    def fromConfig: Option[String] =
+      Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption
+        .orElse(Try(ConfigFactory.load()).toOption)
+        .flatMap(c => Try(c.getConfig("python").getString("path")).toOption)
+        .map(_.trim)
+        .filter(_.nonEmpty)
+
+    def runnable(exe: String): Boolean =
+      Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption
+        .exists { p =>
+          if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+          else p.exitValue() == 0
+        }
+
+    (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable)
+  }
+
+  private def canImportPandas(python: String): Boolean =
+    Try(
+      new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start()
+    ).toOption.exists { p =>
+      if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false }
+      else p.exitValue() == 0
+    }
 }
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/testsupport/setop/SetRows.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/testsupport/setop/SetRows.scala
new file mode 100644
index 00000000000..a8e99a01efc
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/testsupport/setop/SetRows.scala
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.testsupport.setop
+
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple, TupleLike}
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+import scala.io.Source
+
+/** The two inputs the set operators' parity tests share, fed to the native
+  * executor as tuples and to the export as nullable Arrow-style columns.
+  *
+  * Each side holds a finite value and a null the other side also holds, as
+  * controls, next to a NaN and a zero of each sign. The left holds its NaN
+  * twice, since the engine keeps two NaN rows apart as well.
+  */
+object SetRows {
+  private val schema = Schema().add("k", AttributeType.STRING).add("v", AttributeType.DOUBLE)
+
+  private val leftValues: Seq[java.lang.Double] = Seq(1.0, Double.NaN, 0.0, null, 2.0, Double.NaN)
+  private val rightValues: Seq[java.lang.Double] = Seq(1.0, Double.NaN, -0.0, null, 3.0)
+
+  val left: Seq[Tuple] = leftValues.map(v => Tuple(schema, Array[Any]("a", v)))
+  val right: Seq[Tuple] = rightValues.map(v => Tuple(schema, Array[Any]("a", v)))
+
+  /** The native output, one sorted line per row, written as the driver writes it. */
+  def rendered(rows: Iterator[TupleLike]): Seq[String] =
+    rows.map(_.getFields.map(render).mkString(" ")).toSeq.sorted
+
+  /** Runs `exported` over the same rows and returns what it answered, as above. */
+  def runExport(python: String, exported: String, prefix: String): Seq[String] = {
+    val driver =
+      s"""import pandas as pd, numpy as np
+         |
+         |def col(vals):
+         |    mask = np.array([v is None for v in vals])
+         |    return pd.arrays.FloatingArray(np.array([0.0 if v is None else v for v in vals]), mask)
+         |
+         |in1df = pd.DataFrame({"k": ["a"] * ${leftValues.size}, "v": col(${pyList(leftValues)})})
+         |in2df = pd.DataFrame({"k": ["a"] * ${rightValues.size}, "v": col(${pyList(rightValues)})})
+         |$exported
+         |def render(v):
+         |    return "None" if v is pd.NA else v if isinstance(v, str) else repr(float(v))
+         |for row in sorted(" ".join(render(v) for v in r) for r in out1df.itertuples(index=False)):
+         |    print(row)
+         |""".stripMargin
+    val script = Files.createTempFile(prefix, ".py")
+    script.toFile.deleteOnExit()
+    Files.write(script, driver.getBytes(StandardCharsets.UTF_8))
+    val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start()
+    val out = Source.fromInputStream(process.getInputStream, "UTF-8").mkString
+    process.waitFor(120, TimeUnit.SECONDS)
+    if (process.exitValue() != 0) {
+      throw new AssertionError(s"python said:\n$out\nscript:\n$driver")
+    }
+    out.linesIterator.toSeq
+  }
+
+  private def render(value: Any): String =
+    value match {
+      case null                           => "None"
+      case d: java.lang.Double if d.isNaN => "nan"
+      case other                          => other.toString
+    }
+
+  private def pyList(values: Seq[java.lang.Double]): String =
+    values
+      .map(v => if (v == null) "None" else if (v.isNaN) "np.nan" else v.toString)
+      .mkString("[", ", ", "]")
+}