AHeise commented on code in PR #29895: URL: https://github.com/apache/flink/pull/29895#discussion_r4550968101
########## 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) { Review Comment: `TokenStreamRewriter#replace` rejects `from > to`, negative indices and `to >= size` with `replace: range invalid`. Here `stop == start - 1` (what an empty `ParserRuleContext` gives you) records `replaceStop[start] = start - 1`, and for `start > 0` `getText` then loops forever on `i = replaceStop[i] + 1`. All current call sites pass non-empty contexts, so this is defensive, but a hang is a nasty failure mode. Mirroring ANTLR's guard is a one-liner. ########## 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))); Review Comment: Merging the orphan inserts eagerly loses ANTLR's order when a later replace at the same start absorbs them. ANTLR's replace walk folds the raw prior inserts in reverse (`rop.text = iop.text + rop.text` per insert, in instruction order); the after/before combine only happens in the insert walk, which runs later. Repro, fits the generator in the test: ```java delete(t5, t6); insertAfter(t4, "A"); delete(t6, t7); // merges with the first delete, A becomes an orphan at 5 insertAfter(t4, "B"); // resolves the orphan: combine -> "AB" replace(t5, t8, "R"); // folds the combined "AB" ``` `TokenStreamRewriter` emits `BAR`, `FlinkTokenStreamRewriter` emits `ABR`. Only insertAfter is affected; with insertBefore both give `BA`. Unlikely from the splitter, but it breaks the "same output" contract. Keeping the raw list and only marking the index as resolved (combine in `getText`, reverse-fold in `replace`) should fix it. ########## 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: Nit: ANTLR reports this from `getText`; here it is thrown from `replace`/`delete` after the inserts at `start` were already folded and cleared and the ones inside the range dropped, so the rewriter is left half-applied. Fine for the splitter since nothing catches it, but the class Javadoc could say it fails on the edit rather than in `getText`. ########## 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: The generator could be widened a bit: - spans are capped at 4 tokens, while the splitter mostly replaces whole bodies that already contain edits; - EOF is never touched (`size - 2`), so `insertBefore`/`insertAfter` on EOF and the trailing-insert path in `getText` aren't covered; - only one `getText` per sequence, while `BlockStatementGrouper` interleaves `getText(interval)` with further edits. Tracing ANTLR's in-place mutation, I believe outputs stay identical unless an error is thrown, but a randomized check is cheaper than convincing a reviewer of that. The 100k seeds didn't hit the case from the other comment, although it fits the current generator. -- 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]
