chore: Refactor operator serde - part 1

[andygrove]

Which issue does this PR close?

Preparation for fixing #2739 and #2737

Rationale for this change

Partial refactoring of operator serde to use the new framework, similar to what we have done for expressions

What changes are included in this PR?

Created new classes containing operator serde logic. There are no functional changes to this code. It has just been moved out of QueryPlanSerde. This was entirely manual effort - no AI was used.

Note that operator2proto still has the same match statement and simply delegates to these classes. I am working on using a map lookup approach in #2736 but it has exposed some issues such as #2737 that will take time to figure out.

• CometAggregate
• CometExpand
• CometFilter
• CometGlobalLimit
• CometHashJoin
• CometLocalLimit
• CometNativeScan
• CometSortMergeJoin
• CometWindow

Moved some traits out of QueryPlanSerde into their own files. No code changes made.

• CometOperatorSerde
• CometExpressionSerde
• CometAggregateExpressionSerde
• CometScalarFunction

How are these changes tested?

Existing tests

[andygrove]

The diff for the changes to QueryPlanSerde is quite large, but the new code is:

    op match {

      // Fully native scan for V1
      case scan: CometScanExec if scan.scanImpl == CometConf.SCAN_NATIVE_DATAFUSION =>
        CometNativeScan.convert(scan, builder, childOp: _*)

      case filter: FilterExec if CometConf.COMET_EXEC_FILTER_ENABLED.get(conf) =>
        CometFilter.convert(filter, builder, childOp: _*)

      case limit: LocalLimitExec if CometConf.COMET_EXEC_LOCAL_LIMIT_ENABLED.get(conf) =>
        CometLocalLimit.convert(limit, builder, childOp: _*)

      case globalLimitExec: GlobalLimitExec
          if CometConf.COMET_EXEC_GLOBAL_LIMIT_ENABLED.get(conf) =>
        CometGlobalLimit.convert(globalLimitExec, builder, childOp: _*)

      case expand: ExpandExec if CometConf.COMET_EXEC_EXPAND_ENABLED.get(conf) =>
        CometExpand.convert(expand, builder, childOp: _*)

      case _: WindowExec if CometConf.COMET_EXEC_WINDOW_ENABLED.get(conf) =>
        withInfo(op, "Window expressions are not supported")
        None

      case aggregate: HashAggregateExec if CometConf.COMET_EXEC_AGGREGATE_ENABLED.get(conf) =>
        CometHashAggregate.convert(aggregate, builder, childOp: _*)

      case aggregate: ObjectHashAggregateExec
          if CometConf.COMET_EXEC_AGGREGATE_ENABLED.get(conf) =>
        CometObjectHashAggregate.convert(aggregate, builder, childOp: _*)

      case join: BroadcastHashJoinExec
          if CometConf.COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED.get(conf) =>
        CometBroadcastHashJoin.convert(join, builder, childOp: _*)

      case join: ShuffledHashJoinExec if CometConf.COMET_EXEC_HASH_JOIN_ENABLED.get(conf) =>
        CometShuffleHashJoin.convert(join, builder, childOp: _*)

      case join: SortMergeJoinExec =>
        if (CometConf.COMET_EXEC_SORT_MERGE_JOIN_ENABLED.get(conf)) {
          CometSortMergeJoin.convert(join, builder, childOp: _*)
        } else {
          withInfo(join, "SortMergeJoin is not enabled")
          None
        }

[codecov-commenter]

Codecov Report

❌ Patch coverage is 61.92771% with 158 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.45%. Comparing base (f09f8af) to head (812dd7f).
⚠️ Report is 676 commits behind head on main.

Files with missing lines	Patch %	Lines
...ain/scala/org/apache/comet/serde/CometWindow.scala	0.00%	49 Missing ⚠️
...la/org/apache/comet/serde/CometSortMergeJoin.scala	60.60%	14 Missing and 12 partials ⚠️
.../scala/org/apache/comet/serde/CometAggregate.scala	69.62%	15 Missing and 9 partials ⚠️
...scala/org/apache/comet/serde/CometNativeScan.scala	82.17%	9 Missing and 9 partials ⚠️
...n/scala/org/apache/comet/serde/CometHashJoin.scala	68.75%	7 Missing and 8 partials ⚠️
.../scala/org/apache/comet/serde/QueryPlanSerde.scala	56.52%	1 Missing and 9 partials ⚠️
...ain/scala/org/apache/comet/serde/CometExpand.scala	64.28%	4 Missing and 1 partial ⚠️
...cala/org/apache/comet/serde/CometGlobalLimit.scala	50.00%	3 Missing and 1 partial ⚠️
...scala/org/apache/comet/serde/CometLocalLimit.scala	55.55%	3 Missing and 1 partial ⚠️
...ain/scala/org/apache/comet/serde/CometFilter.scala	77.77%	1 Missing and 1 partial ⚠️
... and 1 more

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #2738      +/-   ##
============================================
+ Coverage     56.12%   58.45%   +2.33%     
- Complexity      976     1456     +480     
============================================
  Files           119      159      +40     
  Lines         11743    13892    +2149     
  Branches       2251     2375     +124     
============================================
+ Hits           6591     8121    +1530     
- Misses         4012     4570     +558     
- Partials       1140     1201      +61     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[comphead]

Awesome, thanks @andygrove

[andygrove]

Thanks @comphead! Will merge this and rebase the next one, which fixes the bug you ran into with disabling WindowExec via config.