snuyanzin commented on code in PR #29895: URL: https://github.com/apache/flink/pull/29895#discussion_r4552377930
########## flink-table/flink-table-code-splitter/src/test/java/org/apache/flink/table/codesplit/FlinkTokenStreamRewriterTest.java: ########## @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.codesplit; + +import org.antlr.v4.runtime.CharStreams; +import org.antlr.v4.runtime.CommonTokenStream; +import org.antlr.v4.runtime.Token; +import org.antlr.v4.runtime.TokenStreamRewriter; +import org.antlr.v4.runtime.misc.Interval; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.function.Supplier; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link FlinkTokenStreamRewriter}. */ +class FlinkTokenStreamRewriterTest { + + private static final String CODE = + "class A { int a = 1; void f(int x) { if (x > 0) { a = x; } else { return; } } }"; + + private static final String[] KINDS = {"insertBefore", "insertAfter", "replace", "delete"}; + + private static final int SEEDS = 100_000; + private static final int CHUNKS = 8; + + @ParameterizedTest + @MethodSource("chunks") + @Execution(ExecutionMode.CONCURRENT) + void sameOutputAsTokenStreamRewriter(int chunk) { + for (long seed = (long) chunk * SEEDS / CHUNKS; + seed < (long) (chunk + 1) * SEEDS / CHUNKS; + seed++) { + Random random = new Random(seed); + CommonTokenStream tokens = + new CommonTokenStream(new JavaLexer(CharStreams.fromString(CODE))); + // fills the token stream, so it goes before tokens.size() + FlinkTokenStreamRewriter actual = new FlinkTokenStreamRewriter(tokens); + TokenStreamRewriter expected = new TokenStreamRewriter(tokens); + int size = tokens.size(); + + List<Runnable> actualOps = new ArrayList<>(); + List<Runnable> expectedOps = new ArrayList<>(); + StringBuilder description = new StringBuilder(); + for (int i = random.nextInt(8); i >= 0; i--) { + int start = random.nextInt(size - 1); + Token a = tokens.get(start); + Token b = tokens.get(Math.min(start + random.nextInt(4), size - 2)); Review Comment: I adapted test and making it running with random offset, failure will output problem seed ########## flink-table/flink-table-code-splitter/src/main/java/org/apache/flink/table/codesplit/FlinkTokenStreamRewriter.java: ########## @@ -0,0 +1,242 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.codesplit; + +import org.antlr.v4.runtime.CommonTokenStream; +import org.antlr.v4.runtime.Token; +import org.antlr.v4.runtime.misc.Interval; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * Replacement for {@link org.antlr.v4.runtime.TokenStreamRewriter} with the same output for the + * operations used by the code splitter. {@code TokenStreamRewriter#getText} is quadratic in the + * number of edits, here edits are applied when recorded and {@link #getText} is a single pass. + * + * <p>As in {@code TokenStreamRewriter}: inserts at the start of a replace recorded before it become + * part of its text, a replace drops earlier replaces it contains and earlier inserts strictly + * inside it, overlapping deletes are merged, other overlaps throw, inserts into a replaced range + * fail {@link #getText} unless a later replace drops it, inserts at the start of a replace are + * emitted before it in reverse order. + * + * <p>Unlike {@code TokenStreamRewriter}, {@link #getText} does not change the recorded edits, so + * repeated calls match its first call, and nothing is printed to stdout on merged deletes. + */ +final class FlinkTokenStreamRewriter { + + private final CommonTokenStream tokens; + private final int size; + // replace starting at a token: its last token index (-1 if none) and text (null for delete) + private final int[] replaceStop; + private final String[] replaceText; + // start of the replace covering a token, -1 if none + private final int[] coveredBy; + // inserts before a token in instruction order, the last slot is after EOF + private final List<List<Insert>> inserts; + // starts of replaces with an insert inside them, TokenStreamRewriter fails on these in getText + private final Set<Integer> replacesWithInserts = new HashSet<>(); + // starts of merged deletes with inserts TokenStreamRewriter keeps apart, it fails on these in + // getText unless a later insert there combines with them + private final Set<Integer> orphanInserts = new HashSet<>(); + + FlinkTokenStreamRewriter(CommonTokenStream tokens) { + tokens.fill(); + this.tokens = tokens; + this.size = tokens.size(); + this.replaceStop = new int[size]; + this.replaceText = new String[size]; + this.coveredBy = new int[size]; + this.inserts = new ArrayList<>(Collections.nCopies(size + 1, null)); + Arrays.fill(replaceStop, -1); + Arrays.fill(coveredBy, -1); + } + + void insertBefore(Token token, String text) { + insert(token.getTokenIndex(), text, false); + } + + void insertAfter(Token token, String text) { + insert(token.getTokenIndex() + 1, text, true); + } + + void replace(Token token, String text) { + replace(token, token, text); + } + + void replace(Token start, Token stop, String text) { + replace(start.getTokenIndex(), stop.getTokenIndex(), text); + } + + void delete(Token start, Token stop) { + replace(start.getTokenIndex(), stop.getTokenIndex(), null); + } + + String getText() { + return getText(0, size - 1); + } + + String getText(Interval interval) { + return getText(interval.a, interval.b); + } + + private void insert(int index, String text, boolean after) { + if (index < size && coveredBy[index] != -1 && coveredBy[index] != index) { + // fails getText unless a later replace drops it, kept for a replace starting here + replacesWithInserts.add(coveredBy[index]); + } + if (inserts.get(index) == null) { + inserts.set(index, new ArrayList<>(1)); + } + inserts.get(index).add(new Insert(text, after)); + if (orphanInserts.remove(index)) { + Insert combined = new Insert(combine(inserts.get(index)), after); + inserts.set(index, new ArrayList<>(Collections.singletonList(combined))); + } + } + + private void replace(int start, int stop, String text) { + // as in TokenStreamRewriter, earlier inserts at start become part of the replace text + if (inserts.get(start) != null) { + StringBuilder sb = new StringBuilder(); + appendInsertsReversed(sb, start); + text = text == null ? sb.toString() : sb.append(text).toString(); + inserts.set(start, null); + orphanInserts.remove(start); + } + // only inserts inside this range are dropped, not inside a delete it merges with + for (int i = start + 1; i <= stop; i++) { + inserts.set(i, null); + if (!orphanInserts.isEmpty()) { + orphanInserts.remove(i); + } + } + record(start, stop, text); + } + + private void record(int start, int stop, String text) { + for (int i = start; i <= stop; i++) { + int prevStart = coveredBy[i]; + if (prevStart == -1) { + continue; + } + int prevStop = replaceStop[prevStart]; + boolean bothDeletes = text == null && replaceText[prevStart] == null; + if (prevStart >= start && prevStop <= stop) { + dropReplace(prevStart); + } else if (bothDeletes) { + dropReplace(prevStart); + if (prevStart < start && inserts.get(prevStart) != null) { + orphanInserts.add(prevStart); + } + record(Math.min(start, prevStart), Math.max(stop, prevStop), null); + return; + } else { + throw new IllegalArgumentException( Review Comment: done -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
