[CALCITE-7836] Array functions return incorrect results for ROW elements

[tkalkirill]

Jira Link

CALCITE-7836

Changes Proposed

Fix value-based comparison of ROW elements in array functions, including nested values.

SELECT ARRAY_CONTAINS(ARRAY[ROW(x, y)], ROW(z, w))
FROM (VALUES
  (1, 'Foo', 1, 'Foo'),
  (2, 'Bar', 2, 'Bar')
) AS t(x, y, z, w);

[mihaibudiu]

We prefer quidem tests to jdbc tests (in .iq files), they are much easier to read and write.
Maybe you can convert this one.

[tkalkirill]

Okay, I've moved it to row-equality.iq

[mihaibudiu]

I am not sure what "keys" are.
This really creates sets.
I would not call the elements of a set "keys".
Why does the order matter?

[tkalkirill]

Agreed, “keys” is misleading here; I'll use “set” and “elements”. The documentation (find ARRAY_DISTINCT) explicitly states that ARRAY_DISTINCT preserves element order. The previous implementation also used LinkedHashSet, so this retains the existing behavior.

[mihaibudiu]

please use the same title as the JIRA issue.

[mihaibudiu]

Did you run this on another engine to validate results?
If yes, please mention it in a comment.

[mihaibudiu]

These are simple tests, but it's still not easy for a human to check.

[tkalkirill]

Split the test into separate functions and cases.
I had the check-up done here: https://www.sparkplayground.com/pyspark-online-compiler

def run_validation(display):
    from pyspark.sql import SparkSession, Row
    spark = SparkSession.builder.appName('CALCITE-7836 full results').getOrCreate()
    print('Apache Spark version:', spark.version)
    # Equivalent to the 32 CALCITE-7836 Quidem queries.
    # ARRAY[...] becomes array(...); ROW(...) becomes named_struct(...).
    cases = [
     ('flat', "named_struct('id', x, 'label', y)", "named_struct('id', z, 'label', w)",
      (1, 'Foo'), (2, 'Bar'), (2, 'Car')),
     ('nested_row', "named_struct('r', named_struct('id', x, 'label', y), 'n', 10)", "named_struct('r', named_struct('id', z, 'label', w), 'n', 10)",
      ((1, 'Foo'), 10), ((2, 'Bar'), 10), ((2, 'Car'), 10)),
     ('array_in_row', "named_struct('r', array(named_struct('id', x, 'label', y)), 'n', 10)", "named_struct('r', array(named_struct('id', z, 'label', w)), 'n', 10)",
      ([(1, 'Foo')], 10), ([(2, 'Bar')], 10), ([(2, 'Car')], 10)),
     ('null_field', "named_struct('id', x, 'v', cast(null as int))", "named_struct('id', z, 'v', cast(null as int))",
      (1, None), (2, None), (2, None))
    ]
    def normalize(value):
        if isinstance(value, Row):
            return tuple(normalize(v) for v in value)
        if isinstance(value, list):
            return [normalize(v) for v in value]
        return value
    
    def format_value(value):
        if value is None:
            return 'null'
        if isinstance(value, bool):
            return str(value).lower()
        if isinstance(value, tuple):
            return '{' + ', '.join(format_value(v) for v in value) + '}'
        if isinstance(value, list):
            return '[' + ', '.join(format_value(v) for v in value) + ']'
        return str(value)
    
    summary = []
    for name, a, b, first, left, right in cases:
        equal = left == right
        checks = [
          ('ARRAY_CONTAINS', f'array_contains(array({a}), {b})', [True, equal]),
          ('ARRAY_POSITION', f'array_position(array({a}), {b})', [1, 1 if equal else 0]),
          ('ARRAY_REMOVE', f'array_remove(array({a}, {b}), {b})', [[], [] if equal else [left]]),
          ('ARRAY_DISTINCT', f'array_distinct(array({a}, {b}))', [[first], [left] if equal else [left, right]]),
          ('ARRAY_UNION', f'array_union(array({a}), array({b}))', [[first], [left] if equal else [left, right]]),
          ('ARRAY_INTERSECT', f'array_intersect(array({a}), array({b}))', [[first], [left] if equal else []]),
          ('ARRAY_EXCEPT', f'array_except(array({a}), array({b}))', [[], [] if equal else [left]]),
          ('ARRAYS_OVERLAP', f'arrays_overlap(array({a}), array({b}))', [True, equal])
        ]
        for function, expression, expected in checks:
            query = f"SELECT {expression} AS r FROM VALUES (1, 'Foo', 1, 'Foo'), (2, 'Bar', 2, 'Car') AS t(x, y, z, w) ORDER BY x"
            actual = [normalize(row.r) for row in spark.sql(query).collect()]
            assert actual == expected, f'{name} {function}: expected {expected}, got {actual}'
            summary.append((name, function, format_value(actual[0]), format_value(actual[1]), 'PASS'))
        print(name + ': PASS (8 queries, 16 full results)')
    print('ALL 32 QUERIES / 64 FULL RESULTS MATCH.')
    for name, function, first, second, status in summary:
        print(f'{name} {function}: {first} | {second} [{status}]')
    display(spark.createDataFrame(summary, ['scenario', 'function', 'first_result', 'second_result', 'status']))
    
run_validation(display)

[mihaibudiu]

this looks expensive. Isn't it more efficient if you iterate on the list in the outside loop?

[tkalkirill]

I only moved this code out of the comparer implementation; the algorithm is unchanged.
I don't see a simple way to avoid O(n²) with the current approach while preserving one-to-one matching.
We could avoid shifting elements when removing a match from unmatched, but I would expect the benefit to be limited.
Another option is to use deep-equality wrappers for complete map entries and count their occurrences in a hash map. This would preserve entries with deeply equal keys and give expected O(n) time.
Would you prefer that optimization in this PR, or should we address it in a separate ticket?

[mihaibudiu]

Maybe you can file a ticket for this.

[tkalkirill]

https://issues.apache.org/jira/browse/CALCITE-7841

[mihaibudiu]

This looks fine, but I left a few comments.
If you address them you can also squash the commits.

[mihaibudiu]

I would say "assumes values are immutable".

[mihaibudiu]

Can you add a comment saying that these results were checked using a custom python script which ran the queries against spark? We've had incorrect expected results checked in in the past. I would do this for every function in this file.

[tkalkirill]

I tried to fix this.

[mihaibudiu]

this is an implementation detail, it does not have to be in the test

[mihaibudiu]

same request as above: if these were validated, write this information here.

[tkalkirill]

I tried to fix this.

[mihaibudiu]

Maybe you can file a ticket for this.

[sonarqubecloud]

Quality Gate passed

Issues
3 New issues
0 Accepted issues

Measures
0 Security Hotspots
97.9% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud