Translate PPL dedup Command Part 1: allowedDuplication=1

[LantaoJin]

Description

Syntax

dedup [int] <field-list> keepempty=<bool>] [consecutive=<bool>]

This PR translates following three PPL dedup commands to different Spark Logical Plans.
Note, allowed duplication in this PR is 1. Assuming the <field-list> is a, b.

1. keepempty=false: | dedup [1] a, b [keepempty=false]

Deduplicate ['a, 'b]
+- Filter (isnotnull('a) AND isnotnull('b))
     +- Project
          +- UnresolvedRelation

2. keepempty=true : | dedup [1] a, b keepempty=true

Union
:- Deduplicate ['a, 'b]
:   +- Filter (isnotnull('a) AND isnotnull('b))
:        +- Project
:             +- UnresolvedRelation
+- Filter (isnull('a) OR isnull('b))
     +- Project
          +- UnresolvedRelation

3. consecutive=true: | dedup [1] a, b [keepempty=true] consecutive=true

UnsupportedOperationException("Consecutive deduplication is not supported")

Issues Resolved

Resolves #523 (Subtask of #421)

Check List

• Updated documentation (ppl-spark-integration/README.md)
• Implemented unit tests
• Implemented tests for combination with other commands
• Commits are signed per the DCO using --signoff

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[LantaoJin]

Part 2

To translate dedup command with allowedDuplication > 1, such as | dedup 2 a,b to Spark plan, the solution in my mind is translating to a plan with Window function (e.g row_number) and a new column row_number_col as Filter.

• For | dedup 2 a, b keepempty=false

Project ['a, 'b] // ensure the row_number_col is removed. not a and b only, should contain all fields except row_number_col
+- Filter ('row_number_col <= 2) // allowed duplication = 2
   +- Window [row_number() windowspecdefinition('a, 'b, 'order_key ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS 'row_number_col], ['a, 'b], ['order_key ASC NULLS FIRST]
       +- Filter (isnotnull('a) AND isnotnull('b)) // keepempty=false
          +- Project
             +- UnresolvedRelation

• For | dedup 2 a, b keepempty=true

Union
:- Project ['a, 'b]
:  +- Filter ('row_number_col <= 2)
:     +- Window [row_number() windowspecdefinition('a, 'b, 'order_key ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS 'row_number_col], ['a, 'b], ['order_key ASC NULLS FIRST]
:        +- Filter (isnotnull('a) AND isnotnull('b))
:           +- Project
:              +- UnresolvedRelation
+- Filter (isnull('a) OR isnull('b))
   +- Project
      +- UnresolvedRelation

One un-clarified point is what is the 'order_key in Window node?

Option 1

AFAIK, there always should be a time series field existing for log analysis, such as _time or @timestamp. How to get it in AST?

Option 2

If we cannot point out an existing time series field as order_key, an option is change the the syntax of dedup command as following (make sure sql repo will change first)

| dedup 2 a, b order by c
| dedup 2 a, b orderkey=c

Option 3

partitioning and ordering by the same columns:

Window.partitionBy(dedupColumns).orderBy(dedupColumns)

If that, the results can be non-deterministic for large datasets. This non-determinism arises because the sorting does not have a secondary tie-breaking rule to ensure a consistent order of rows within each partition. If multiple rows have the same values for the partition and order columns, their relative order can vary across different runs

A complex case for option 3 is while we have a sorter before dedup command

| sort c | dedup 2 a,b

Any thoughts? @YANG-DB @penghuo @dai-chen

FYI: In Spark, window function requires window to be ordered. Query fails if order by clause is not specified. For example, SELECT <wf>(expr) OVER ([PARTITION BY window_partition] ORDER BY window_ordering) from table
But in MySQL or opensearch-sql, the order by clause can be omitted.

[LantaoJin]

The second un-clarified point is about node Deduplicate and DeduplicateWithinWatermark
https://issues.apache.org/jira/browse/SPARK-42931 introduced a new LogicalPlan DeduplicateWithinWatermark to deduplicate events within the watermark and was available in 3.5.0+
I am not clear the architecture of Flint, but seems it performs on Spark Streaming. Deduplication on streaming dataset has different semantic with batch dataset:

We document the behavior clearly that the event time column should be a part of the subset columns for deduplication to clean up the state, but it cannot be applied to the customers as timestamps are not exactly the same for duplicated events in their use cases.

Need more guidance on the purpose of this repository and how it should be used in production or in future. For current time-being, I choose Deduplicate in this PR (DeduplicateWithinWatermark required 3.5.0+).

[penghuo]

Part 2

To translate dedup command with allowedDuplication > 1, such as | dedup 2 a,b to Spark plan, the solution in my mind is translating to a plan with Window function (e.g row_number) and a new column row_number_col as Filter.

• For | dedup 2 a, b keepempty=false

Project ['a, 'b] // ensure the row_number_col is removed. not a and b only, should contain all fields except row_number_col
+- Filter ('row_number_col <= 2) // allowed duplication = 2
   +- Window [row_number() windowspecdefinition('a, 'b, 'order_key ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS 'row_number_col], ['a, 'b], ['order_key ASC NULLS FIRST]
       +- Filter (isnotnull('a) AND isnotnull('b)) // keepempty=false
          +- Project
             +- UnresolvedRelation

• For | dedup 2 a, b keepempty=true

Union
:- Project ['a, 'b]
:  +- Filter ('row_number_col <= 2)
:     +- Window [row_number() windowspecdefinition('a, 'b, 'order_key ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS 'row_number_col], ['a, 'b], ['order_key ASC NULLS FIRST]
:        +- Filter (isnotnull('a) AND isnotnull('b))
:           +- Project
:              +- UnresolvedRelation
+- Filter (isnull('a) OR isnull('b))
   +- Project
      +- UnresolvedRelation

One un-clarified point is what is the 'order_key in Window node? AFAIK, there always should be a time series field existing for log analysis, such as _time or @timestamp. How to get it in AST? If we cannot point out an existing time series field as order_key, the syntax of dedup command might change to

| dedup 2 a, b order by c
| dedup 2 a, b orderkey=c

Any thoughts? @YANG-DB @penghuo @dai-chen

FYI: In Spark, window function requires window to be ordered. Query fails if order by clause is not specified. For example, SELECT <wf>(expr) OVER ([PARTITION BY window_partition] ORDER BY window_ordering) from table But in MySQL or opensearch-sql, the order by clause can be omitted.

Agree on Windows function.

Questions

• What is the difference of Deduplicate and Window? Does Deduplicate will be optimized as WIndows function?
• Can we windows_partition and window_ordering be same field?

[LantaoJin]

Questions

What is the difference of Deduplicate and Window? Does Deduplicate will be optimized as WIndows function?
Can we windows_partition and window_ordering be same field?

1. They are totally different. Actually, Deduplicate will be optimized as Aggregation. Using Deduplicate instead of Aggregation in code is more readable. Ref https://github.com/apache/spark/blob/b3877894acb512477d41038269f20158d7bca8f3/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L2261
2. Yes we can. But it brings non-deterministic results, order can vary across different runs.

I updated the comment #521 (comment)

[LantaoJin]

Summary

Discussed with @penghuo offline. Here is the conclusions:

1. For node Deduplicate and DeduplicateWithinWatermark

Deduplicate will be optimized as Aggregation, but DeduplicateWithinWatermark will be transformed to a new physical plan node StreamingDeduplicateWithinWatermarkExec. Considering current usage, #521 will keep using Deduplicate.

2. For the 'order_key in Window

Will choose the option 3 (partitioning and ordering by the same columns). Ref #521 (comment)

CC @YANG-DB and @dai-chen

Again, all above shouldn't block code review for this PR.

[YANG-DB]

not mandatory but for future - lets extract the concrete login into a dedicated strategy class and inject that into the QueryPlanner
this would reduce the complexity and simplify the testing and seperation of concerns

[LantaoJin]

@penghuo @dai-chen @YANG-DB Any other concern on it? Would this PR be merged now?