[FLINK-27246][TableSQL/Runtime][master] - Include WHILE blocks in split generated java methods

[kristoffSC]

What is the purpose of the change

This PR fix issue reported in https://issues.apache.org/jira/browse/FLINK-27246 that was caused by JavaCodeSplitter not rewriting "while" statements that was leading to Java compiler not able to compile code due to to big methods.

Proposed change, enhance JavaCodeSplitter by replacing IfStatementRewriter.java with BlockStatementRewriter.java.

The new BlockStatementRewriter.java can rewrite both IF/ESLE statements and WHILE blocks including nested statements.

The BlockStatementRewriter works in two steps. First step extracts blocks from IF/ELSE branches and WHILE bodies to separate methods. This step is executed by new class BlockStatementSplitter.

The second step, groups statements (single lien statements, IF/ELSE and WHILE statements) into groups and extract them to new methods. This step is executed by new class BlockStatementGrouper .

An example of rewritten code:
BEFORE:

public class Example {
 
      int b;
      int c;
 
      public void myFun(int a) {
 
          int counter = 10;
          while (counter > 0) {
              int localA = a + 1000;
              System.out.println(localA);
              if (a > 0) {
                  b = a * 2;
                  c = b * 2;
                  System.out.println(b);
              } else {
                  b = a * 3;
                  System.out.println(b);
              }
              counter--;
          }
      }
  }

AFTER executing BlockStatementRewriter:

public class Example {
 
      int b;
      int c;
 
      public void myFun(int a) {
 
          int counter = 10;
 
          while (counter > 0) {
              myFun_rewriteGroup1(a);
              counter--;
          }
      }
 
      void myFun_rewriteGroup1(int a) {
          myFun_whileBody0_0(a);
          if (a > 0) {
              myFun_whileBody0_0_ifBody0(a);
          } else {
              myFun_whileBody0_0_ifBody1(a);
          }
      }
 
      void myFun_whileBody0_0(int a) {
          int localA = a + 1000;
          System.out.println(localA);
      }
 
      void myFun_whileBody0_0_ifBody1(int a) {
          b = a * 3;
          System.out.println(b);
      }
 
      void myFun_whileBody0_0_ifBody0(int a) {
          b = a * 2;
          c = b * 2;
          System.out.println(b);
      }
  }

Brief change log

• Include WHILE blocs in Java cod splitter logic.
• Add BlockStatementRewriter, BlockStatementSplitter and BlockStatementGrouper class,
• Remove IfStatementRewriter class

Verifying this change

This change is already covered by existing tests, such as

• Flink-brach-sql-test
• Flink-tcpds-test
• Junit test - JavaCodeSplitterTest

This change added tests and can be verified as follows:

• BlockStatementRewriterTest.java
• BlockStatementSplitterTest.java
• BlockStatementGrouperTest.java

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[flinkbot]

CI report:

• ac87f84 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@flinkbot run azure

[kristoffSC]

@tsreaper

I've replayed to/implemented changes from your last comments.

Regarding I'd like to see more examples. Could you please provide examples and explain

If we focus only on IF/ELSE statements without else if blocks which are also supported in proposed solution I think that the differences boils down to one example below.

Having:

public void myFun1(int[] a, int[] b) throws RuntimeException {
        if (a[0] == 0) {
            a[11] = b[0];
            a[12] = b[0];
            if (a[2] == 0) {
                a[21] = 1;
                a[22] = 1;
            } else {
                a[23] = b[2];
                a[24] = b[2];
            }

            a[13] = b[0];
            a[14] = b[0];
        }
}

The original `IfStatementRewritter will not extract:

            a[11] = b[0];
            a[12] = b[0];

and

            a[13] = b[0];
            a[14] = b[0];

To their's separate methods. They will be extracted together with entire TRUE branch code block. In my proposed solution, the TRUE branch will be extracted plus statements from above will be further extracted to their own methods like so:

    void myFun1_0_1(int[] a, int[] b) throws RuntimeException {
        a[11] = b[0];
        a[12] = b[0];
    }

    void myFun1_0(int[] a, int[] b) throws RuntimeException {
        a[13] = b[0];
        a[14] = b[0];
    }

The test with similar code is implemented in: BlockStatementRewriterTest::testIfMultipleSingleLineStatementRewrite

If you would have 3rd level IF/ELSE after a[21] = 1; a[22] = 1; the story will be the same. My proposition is able to extract such statements for every level.

And now you may say, that this is not a big deal. Maybe with an example I'm showing here the gain is hard to spot. However I can tell that for a production job (SQL Query), that was causing FLINK-27246, I have an extracted method that has 537 statements similar to those from example below, that I'm not sure if the original implementation could extract. Plus we have many additional methods containing 2 statements.

So even if this would be the only extra thing (except supporting else if and while) I think that at scale, the gain adds up.

[tsreaper]

Thanks @kristoffSC for your effort in this long running pull request. I don't have much to say now. You did a great job.

One last comment: Add a test to CodeSplitITCase to address why this issue is brought up. Maciej Bryński's comment in the original JIRA ticket might be useful.

[kristoffSC]

Hi @tsreaper
thanks for kind words. I'm really happy that we are getting closer to merge this change.

I was using Maciej Bryński's SQL example from FLINK-27246 ticket to verify if my change here solves the original problem.
I had Flink Job (TableAPI) that was running with code splitter changes we have here.

I will add it to CodeSplitITCase in following days.

[kristoffSC]

Hi @tsreaper
I've implemented test you requested for. It's 'CodeSplitITCase ::testManyAggregationsWithGroupBy'

CI/CD is green.

Anything else you would like me to do? If not could you merge this PR?

P.S.
After merging this PR I would like to create a back port PRs to 1.15 and 1.16. Our client is on 1.14 which is not supported anymore and I'm not sure to which version our client will be upgrading to.

[tsreaper]

After merging this PR I would like to create a back port PRs to 1.15 and 1.16.

You're welcome to create back ports. As 1.16.0 has just released for one month or two I guess 1.16.1 will come in the near future. However 1.15 is an older version and I can't determine when will we release the next version (or if the next version will ever come).

[snuyanzin]

it looks like it generates methods with same signature for some cases e.g. org.apache.flink.table.planner.runtime.stream.sql.MatchRecognizeITCase#testUserDefinedFunctions
and it leads to debug output for these tests

[kristoffSC]

@snuyanzin

it looks like it generates methods with same signature for some cases e.g.

I will take a look at this one.

This was not causing CI build to fail?
Im guessing not, since in case of compilation error on rewritten code, planner is using original the one.. hm.

[snuyanzin]

somehow build doesn't fail
however lots of output in tests with generated code
The output happens here

flink/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/generated/CompileUtils.java

Line 106 in 2ae5df2

System.out.println(addLineNumber(code));

[snuyanzin]

btw, i created an issue for that
https://issues.apache.org/jira/browse/FLINK-30927

[kristoffSC]

I already have fix, will provide PR shortly.