[kz930]

What changes were proposed in this PR?

Distinct's exported Python called drop_duplicates, which counts two NaN values as equal and 0.0 as equal to -0.0. The engine's DistinctOpExec keeps rows in a LinkedHashSet that compares them with Scala's ==, under which NaN equals nothing, and hashes each double by its bits, so 0.0 and -0.0 land apart. The export therefore emitted fewer rows than the run it came from.

The export now selects rows with _texera_set_select, the helper the set operators use: one key per row, equal exactly where the engine's rows are, keeping the first row of each key in input order.

Any related issues, documentation, discussions?

Closes #8974. Depends on #8506, which adds StandaloneHelpers.SetRowKeys; until it merges this branch does not compile.

How was this PR tested?

DistinctOpDescSpec gains a test that runs DistinctOpExec and the exported Python over 1.0, NaN, 0.0, -0.0, null, 1.0, NaN, null, -0.0 and compares the rows in order. The engine keeps 6. The old export kept 4 and fails the test. It replaces the test that pinned the old drop_duplicates line. With #8506 applied, all 11 Distinct tests pass, and Distinct's verification run still matches. WorkflowToPythonResourceSpec, which looked for drop_duplicates in the served script, now looks for _texera_set_select.

Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)

🤖 Generated with Claude Code

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• No candidates found from git blame history.

[codecov-commenter]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
22	1	21	0

View the top 1 failed test(s) by shortest run time

org.apache.texera.amber.operator.distinct.DistinctOpDescSpec::DistinctOpDesc.generateStandaloneCode should keep the rows the engine keeps for NaN and signed zero

Stack Traces | 1.49s run time

org.scalatest.exceptions.TestFailedException: python said:
Traceback (most recent call last):
  File ".../hn37q6pj7sj_9dmqqc2113800000gn/T/distinct-nan-zero-10238351285047588606.py", line 133, in <module>
    [_texera_distinct_codes] = _texera_key_codes(in1df, signed=True)
                               ^^^^^^^^^^^^^^^^^
NameError: name '_texera_key_codes' is not defined

script:
import pandas as pd, numpy as np
vals = [1.0, np.nan, 0.0, -0.0, None, 1.0, np.nan, None, -0.0]
mask = np.array([v is None for v in vals])
in1df = pd.DataFrame({"v": pd.arrays.FloatingArray(
    np.array([0.0 if v is None else v for v in vals]), mask)})
# AttributeTypeUtils, transcribed so a cast answers as the engine does.
def _texera_cast_boolean(x):
    # toBoolean first, then `toInt == 1`: "0" and "2" are both false.
    if isinstance(x, str):
        text = x.strip()
        lowered = text.lower()
        if lowered == "true":
            return True
        if lowered == "false":
            return False
        return int(text) == 1
    return x != 0


def _texera_parse_number(text):
    # java.text.NumberFormat for Locale.US, which a cast reaches through
    # `parseField(force = true)`. Lenient where Python's int is not: it stops
    # at the first character that cannot continue a number ("12abc" is 12),
    # drops "," without checking group sizes ("1,23" is 123), and reads "."
    # as a decimal point. It refuses a leading "+" and text with no digits.
    #
    # Also returns whether it came back as a Double, which decides between
    # the two narrowings below.
    import re

    match = re.match(r"\s*(-?)([0-9,]*)(?:\.([0-9]*))?", text)
    sign = -1 if match.group(1) == "-" else 1
    digits = (match.group(2) or "").replace(",", "")
    fraction = match.group(3) or ""
    if not digits and not fraction:
        raise ValueError("Unparseable number: " + repr(text))
    if fraction:
        return sign * float((digits or "0") + "." + fraction), True
    value = sign * int(digits)
    # A whole number past long's range comes back as a Double.
    if -(2 ** 63) <= value <= 2 ** 63 - 1:
        return value, False
    return float(value), True


def _texera_long_value(value, is_double):
    # Number.longValue(): a Double truncates toward zero and saturates at
    # long's bounds, where a Long is already itself.
    if is_double:
        return max(-(2 ** 63), min(2 ** 63 - 1, int(value)))
    return value


def _texera_cast_integral(x):
    # The LONG target. Scala's toLong takes no decimal point of its own, so
    # a fraction only ever arrives already parsed.
    if isinstance(x, str):
        return _texera_long_value(*_texera_parse_number(x))
    if isinstance(x, bool):
        return 1 if x else 0
    if isinstance(x, float):
        return _texera_long_value(x, True)
    return int(x)


def _texera_cast_int32(x, wrap):
    # The two 32-bit narrowings differ: Long.toInt keeps the low bits, so
    # 2147483648 comes back as -2147483648, while Double.toInt saturates.
    #
    # Text decides by what the parse returned; everything else is told by the
    # SOURCE column's declared type, because `Series.apply` hands the cells of
    # a nullable integer column over as floats.
    if isinstance(x, str):
        value, is_double = _texera_parse_number(x)
        wrap = not is_double
    elif isinstance(x, bool):
        value = 1 if x else 0
    else:
        value = int(x)
    if wrap:
        return ((value + 2147483648) % 4294967296) - 2147483648
    # int() first: Double.intValue truncates toward zero before it clamps.
    return max(-2147483648, min(2147483647, int(value)))


def _texera_epoch_millis_to_timestamp(s):
    # `new Timestamp(long)` reads MILLISECONDS where pd.to_datetime defaults
    # to nanoseconds, and renders in the JVM's default zone, so leaving the
    # result in UTC would put it a whole offset away. Text needs neither: a
    # parsed wall clock is already the wall clock.
    # tzlocal() and not the current offset: the zone carries its daylight
    # rules, and each instant needs the offset in force when it happened.
    from dateutil.tz import tzlocal

    return (
        pd.to_datetime(s, unit="ms", errors="coerce", utc=True)
        .dt.tz_convert(tzlocal())
        .dt.tz_localize(None)
    )


def _texera_cast_double(x):
    if isinstance(x, str):
        return float(x.strip())
    return float(x)


def _texera_cast_string(s):
    # `toString` on the field, so the COLUMN's type decides the text and
    # not the shape of the value: a double keeps its point, whether or not
    # it lands on a whole number, and an integer never grows one. A column
    # of no single type is read value by value.
    if pd.api.types.is_bool_dtype(s):
        return s.map(lambda x: None if pd.isna(x) else ("true" if x else "false"))
    if pd.api.types.is_integer_dtype(s):
        return s.map(lambda x: None if pd.isna(x) else str(int(x)))
    if pd.api.types.is_datetime64_any_dtype(s):
        # java.sql.Timestamp.toString: trailing zeros dropped from the
        # fraction, but never all of them. Python's own str() writes no
        # fraction at all on a whole second and six digits otherwise.
        def _ts(x):
            if pd.isna(x):
                return None
            text = x.strftime("%Y-%m-%d %H:%M:%S.%f").rstrip("0")
            return text + "0" if text.endswith(".") else text

        return s.map(_ts)
    return s.map(
        lambda x: None
        if pd.isna(x)
        else ("true" if x else "false") if pd.api.types.is_bool(x) else str(x)
    )
[_texera_distinct_codes] = _texera_key_codes(in1df, signed=True)
out1df = in1df[(~_texera_distinct_codes.duplicated()).to_numpy()].reset_index(drop=True)
for v in out1df["v"]:
    print("None" if v is pd.NA else repr(float(v)))
1 was not equal to 0
      at org.scalatest.matchers.MatchersHelper$.indicateFailure(MatchersHelper.scala:392)
      at org.scalatest.matchers.should.Matchers$AnyShouldWrapper.shouldBe(Matchers.scala:7539)
      at org.apache.texera.amber.operator.distinct.DistinctOpDescSpec.$anonfun$new$13(DistinctOpDescSpec.scala:169)
      at org.scalatest.Assertions.withClue(Assertions.scala:1065)
      at org.scalatest.Assertions.withClue$(Assertions.scala:1052)
      at org.scalatest.flatspec.AnyFlatSpec.withClue(AnyFlatSpec.scala:1686)
      at org.apache.texera.amber.operator.distinct.DistinctOpDescSpec.$anonfun$new$8(DistinctOpDescSpec.scala:168)
      at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
      at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
      at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
      at org.scalatest.Transformer.apply(Transformer.scala:22)
      at org.scalatest.Transformer.apply(Transformer.scala:20)
      at org.scalatest.flatspec.AnyFlatSpecLike$$anon$5.apply(AnyFlatSpecLike.scala:1832)
      at org.scalatest.TestSuite.withFixture(TestSuite.scala:196)
      at org.scalatest.TestSuite.withFixture$(TestSuite.scala:195)
      at org.scalatest.flatspec.AnyFlatSpec.withFixture(AnyFlatSpec.scala:1686)
      at org.scalatest.flatspec.AnyFlatSpecLike.invokeWithFixture$1(AnyFlatSpecLike.scala:1830)
      at org.scalatest.flatspec.AnyFlatSpecLike.$anonfun$runTest$1(AnyFlatSpecLike.scala:1842)
      at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
      at org.scalatest.flatspec.AnyFlatSpecLike.runTest(AnyFlatSpecLike.scala:1842)
      at org.scalatest.flatspec.AnyFlatSpecLike.runTest$(AnyFlatSpecLike.scala:1824)
      at org.scalatest.flatspec.AnyFlatSpec.runTest(AnyFlatSpec.scala:1686)
      at org.scalatest.flatspec.AnyFlatSpecLike.$anonfun$runTests$1(AnyFlatSpecLike.scala:1900)
      at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
      at scala.collection.immutable.List.foreach(List.scala:323)
      at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
      at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:390)
      at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:427)
      at scala.collection.immutable.List.foreach(List.scala:323)
      at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
      at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396)
      at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475)
      at org.scalatest.flatspec.AnyFlatSpecLike.runTests(AnyFlatSpecLike.scala:1900)
      at org.scalatest.flatspec.AnyFlatSpecLike.runTests$(AnyFlatSpecLike.scala:1899)
      at org.scalatest.flatspec.AnyFlatSpec.runTests(AnyFlatSpec.scala:1686)
      at org.scalatest.Suite.run(Suite.scala:1114)
      at org.scalatest.Suite.run$(Suite.scala:1096)
      at org.scalatest.flatspec.AnyFlatSpec.org$scalatest$flatspec$AnyFlatSpecLike$$super$run(AnyFlatSpec.scala:1686)
      at org.scalatest.flatspec.AnyFlatSpecLike.$anonfun$run$1(AnyFlatSpecLike.scala:1945)
      at org.scalatest.SuperEngine.runImpl(Engine.scala:535)
      at org.scalatest.flatspec.AnyFlatSpecLike.run(AnyFlatSpecLike.scala:1945)
      at org.scalatest.flatspec.AnyFlatSpecLike.run$(AnyFlatSpecLike.scala:1943)
      at org.scalatest.flatspec.AnyFlatSpec.run(AnyFlatSpec.scala:1686)
      at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
      at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
      at sbt.TestRunner.runTest$1(TestFramework.scala:153)
      at sbt.TestRunner.run(TestFramework.scala:168)
      at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.$anonfun$apply$1(TestFramework.scala:336)
      at sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:296)
      at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
      at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
      at sbt.TestFunction.apply(TestFramework.scala:348)
      at sbt.Tests$.$anonfun$toTask$1(Tests.scala:436)
      at sbt.std.Transform$$anon$3.$anonfun$apply$2(Transform.scala:47)
      at sbt.std.Transform$$anon$4.work(Transform.scala:69)
      at sbt.Execute.$anonfun$submit$2(Execute.scala:283)
      at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:24)
      at sbt.Execute.work(Execute.scala:292)
      at sbt.Execute.$anonfun$submit$1(Execute.scala:283)
      at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
      at sbt.CompletionService$$anon$2.call(CompletionService.scala:65)
      at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
      at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
      at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
      at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
      at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
      at java.base/java.lang.Thread.run(Thread.java:840)

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 10 worse · ⚪ 5 noise (<±5%) · 0 without baseline

Compared against main 6a29c59 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	372	0.227	26,373/33,008/33,008 us	🔴 +15.8% / 🔴 +193.3%
🔴	bs=100 sw=10 sl=64	800	0.488	121,727/168,004/168,004 us	🔴 +29.8% / 🔴 +124.4%
🔴	bs=1000 sw=10 sl=64	952	0.581	1,043,254/1,218,795/1,218,795 us	🔴 +5.3% / 🔴 +72.5%

Baseline details

Latest main 6a29c59 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	372 tuples/sec	420 tuples/sec	1,119 tuples/sec	-11.4%	-66.8%
bs=10 sw=10 sl=64	MB/s	0.227 MB/s	0.256 MB/s	0.683 MB/s	-11.3%	-66.8%
bs=10 sw=10 sl=64	p50	26,373 us	22,769 us	8,992 us	+15.8%	+193.3%
bs=10 sw=10 sl=64	p95	33,008 us	32,485 us	11,255 us	+1.6%	+193.3%
bs=10 sw=10 sl=64	p99	33,008 us	32,485 us	13,808 us	+1.6%	+139.1%
bs=100 sw=10 sl=64	throughput	800 tuples/sec	864 tuples/sec	1,518 tuples/sec	-7.4%	-47.3%
bs=100 sw=10 sl=64	MB/s	0.488 MB/s	0.527 MB/s	0.927 MB/s	-7.4%	-47.3%
bs=100 sw=10 sl=64	p50	121,727 us	114,884 us	68,504 us	+6.0%	+77.7%
bs=100 sw=10 sl=64	p95	168,004 us	129,435 us	74,866 us	+29.8%	+124.4%
bs=100 sw=10 sl=64	p99	168,004 us	129,435 us	89,939 us	+29.8%	+86.8%
bs=1000 sw=10 sl=64	throughput	952 tuples/sec	962 tuples/sec	1,569 tuples/sec	-1.0%	-39.3%
bs=1000 sw=10 sl=64	MB/s	0.581 MB/s	0.587 MB/s	0.957 MB/s	-1.0%	-39.3%
bs=1000 sw=10 sl=64	p50	1,043,254 us	1,029,443 us	669,501 us	+1.3%	+55.8%
bs=1000 sw=10 sl=64	p95	1,218,795 us	1,157,686 us	706,487 us	+5.3%	+72.5%
bs=1000 sw=10 sl=64	p99	1,218,795 us	1,157,686 us	737,560 us	+5.3%	+65.2%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,536.99,200,128000,372,0.227,26373.24,33007.50,33007.50
1,100,10,64,20,2500.19,2000,1280000,800,0.488,121727.33,168004.11,168004.11
2,1000,10,64,20,21011.36,20000,12800000,952,0.581,1043254.17,1218795.39,1218795.39

[kz930]

@carloea2 could you review this one when you have time? It builds on #8506 (which adds SetRowKeys), so the red CI is expected until that merges. Please test it with #8506 applied.

[kz930]

@carloea2 A friendly ping on this one; I've since added a test expecting Distinct's set select in the export.

[carloea2]

DistinctOpDescSpec passes all 8 tests with the declared SetRowKeys helper supplied. WorkflowToPythonResourceSpec passes all 4 tests, and scalafmtCheckAll passes. NaN, null and signed-zero cases match the native distinct behavior. Keep the draft dependency on #8506.