This is an automated email from the ASF dual-hosted git repository. jhyde pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/calcite.git
commit 267326d15e347df18af816127b5e5cef6a8bf3d4 Author: Julian Hyde <[email protected]> AuthorDate: Fri Jun 2 22:37:52 2023 -0700 [CALCITE-5764] Add Puffin Puffin provides a similar programming model to Awk. A program is a collection of rules, each of which is a predicate applied to a line of text followed by an action on that test. Puffin programs are thread-safe; if Puffin is invoked on mutiple sources (files), each file is processed in its own thread and is allocated its own state that the rules can use without coordination. Puffin caches compiled regular expressions, so that calls to `matches(String)` are almost as efficient as if the program had called `Pattern.compile(String)` in advance. --- .../main/java/org/apache/calcite/util/Puffin.java | 332 +++++++++++++++++++++ .../main/java/org/apache/calcite/util/Sources.java | 18 +- .../java/org/apache/calcite/test/PuffinTest.java | 108 +++++++ 3 files changed, 452 insertions(+), 6 deletions(-) diff --git a/core/src/main/java/org/apache/calcite/util/Puffin.java b/core/src/main/java/org/apache/calcite/util/Puffin.java new file mode 100644 index 0000000000..f776d58030 --- /dev/null +++ b/core/src/main/java/org/apache/calcite/util/Puffin.java @@ -0,0 +1,332 @@ +/* + * 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.calcite.util; + +import org.apache.calcite.runtime.PairList; +import org.apache.calcite.runtime.Unit; + +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.collect.ImmutableList; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.OutputStream; +import java.io.PrintWriter; +import java.io.Reader; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Predicate; +import java.util.function.Supplier; +import java.util.regex.Pattern; +import java.util.stream.Stream; + +import static java.util.Objects.requireNonNull; + +/** + * A text processor similar to Awk. + * + * <p>Example use: + * + * <blockquote><pre>{@code + * File file; + * final Puffin.Program program = + * Puffin.builder() + * .add(line -> !line.startsWith("#"), + * line -> counter.incrementAndGet()) + * .after(context -> + * context.println("There were " + counter.get() + * + " uncommented lines")) + * .build(); + * program.execute(Source.of(file), System.out); + * }</pre></blockquote> + * + * <p>prints the following to stdout: + * + * <blockquote>{@code + * There were 3 uncommented lines. + * }</blockquote> + */ +public class Puffin { + private Puffin() { + } + + /** Creates a Builder. + * + * @param fileStateFactory Creates the state for each file + * @return Builder + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file + */ + public static <G, F> Builder<G, F> builder(Supplier<G> globalStateFactory, + Function<G, F> fileStateFactory) { + return new BuilderImpl<>(globalStateFactory, fileStateFactory, + PairList.of(), new ArrayList<>()); + } + + /** Creates a Builder with no state. */ + public static Builder<Unit, Unit> builder() { + return builder(() -> Unit.INSTANCE, u -> u); + } + + /** Fluent interface for constructing a Program. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file + * @see Puffin#builder */ + public interface Builder<G, F> { + Builder<G, F> add(Predicate<Line<G, F>> linePredicate, + Consumer<Line<G, F>> action); + Builder<G, F> after(Consumer<Context<G, F>> action); + Program<G> build(); + } + + /** A Puffin program. You can execute it on a file. + * + * @param <G> Type of state that is created when we start processing */ + public interface Program<G> { + /** Executes this program. */ + G execute(Stream<? extends Source> sources, PrintWriter out); + + /** Executes this program, writing to an output stream such as + * {@link System#out}. */ + default void execute(Stream<? extends Source> sources, OutputStream out) { + try (PrintWriter w = Util.printWriter(out)) { + execute(sources, w); + } + } + + /** Executes this program on a single source. */ + default void execute(Source source, OutputStream out) { + execute(Stream.of(source), out); + } + } + + /** A line in a file. + * + * <p>Created by an executing program and passed to the predicate + * and action that you registered in + * {@link Builder#add(Predicate, Consumer)}. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file + */ + public interface Line<G, F> { + G globalState(); + F state(); + int fnr(); + Source source(); + boolean startsWith(String prefix); + boolean contains(CharSequence s); + boolean endsWith(String suffix); + boolean matches(String regex); + String line(); + } + + /** Context for executing a Puffin program within a given file. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file */ + public static class Context<G, F> { + final PrintWriter out; + final Source source; + final F fileState; + final G globalState; + private final Function<String, Pattern> patternCache; + + /** Holds the current line. */ + String line = ""; + + /** Holds the current line number in the file (starting from 1). + * + * <p>Corresponds to the Awk variable {@code FNR}, which stands for "file + * number of records". */ + int fnr = 0; + + Context(PrintWriter out, Source source, + Function<String, Pattern> patternCache, G globalState, + F fileState) { + this.out = requireNonNull(out, "out"); + this.source = requireNonNull(source, "source"); + this.patternCache = requireNonNull(patternCache, "patternCache"); + this.globalState = requireNonNull(globalState, "globalState"); + this.fileState = requireNonNull(fileState, "fileState"); + } + + public F state() { + return fileState; + } + + public G globalState() { + return globalState; + } + + public void println(String s) { + out.println(s); + } + + Pattern pattern(String regex) { + return patternCache.apply(regex); + } + } + + /** Extension to {@link Context} that also implements {@link Line}. + * + * <p>We don't want clients to know that {@code Context} implements + * {@code Line}, but neither do we want to create a new {@code Line} object + * for every line in the file. Making this a subclass accomplishes both + * goals. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file */ + static class ContextImpl<G, F> extends Context<G, F> implements Line<G, F> { + + ContextImpl(PrintWriter out, Source source, + Function<String, Pattern> patternCache, G globalState, F state) { + super(out, source, patternCache, globalState, state); + } + + @Override public int fnr() { + return fnr; + } + + @Override public Source source() { + return source; + } + + @Override public boolean startsWith(String prefix) { + return line.startsWith(prefix); + } + + @Override public boolean contains(CharSequence s) { + return line.contains(s); + } + + @Override public boolean endsWith(String suffix) { + return line.endsWith(suffix); + } + + @Override public boolean matches(String regex) { + return pattern(regex).matcher(line).matches(); + } + + @Override public String line() { + return line; + } + } + + /** Implementation of {@link Program}. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file */ + private static class ProgramImpl<G, F> implements Program<G> { + private final Supplier<G> globalStateFactory; + private final Function<G, F> fileStateFactory; + private final PairList<Predicate<Line<G, F>>, Consumer<Line<G, F>>> pairList; + private final ImmutableList<Consumer<Context<G, F>>> endList; + @SuppressWarnings("Convert2MethodRef") + private final LoadingCache<String, Pattern> patternCache0 = + CacheBuilder.newBuilder() + .build(CacheLoader.from(regex -> Pattern.compile(regex))); + private final Function<String, Pattern> patternCache = + patternCache0::getUnchecked; + + private ProgramImpl(Supplier<G> globalStateFactory, + Function<G, F> fileStateFactory, + PairList<Predicate<Line<G, F>>, Consumer<Line<G, F>>> pairList, + ImmutableList<Consumer<Context<G, F>>> endList) { + this.globalStateFactory = globalStateFactory; + this.fileStateFactory = fileStateFactory; + this.pairList = pairList; + this.endList = endList; + } + + @Override public G execute(Stream<? extends Source> sources, + PrintWriter out) { + final G globalState = globalStateFactory.get(); + sources.forEach(source -> execute(globalState, source, out)); + return globalState; + } + + private void execute(G globalState, Source source, PrintWriter out) { + try (Reader r = source.reader(); + BufferedReader br = new BufferedReader(r)) { + final F fileState = fileStateFactory.apply(globalState); + final ContextImpl<G, F> x = + new ContextImpl<G, F>(out, source, patternCache, globalState, + fileState); + for (;;) { + String lineText = br.readLine(); + if (lineText == null) { + endList.forEach(end -> end.accept(x)); + break; + } + ++x.fnr; + x.line = lineText; + pairList.forEach((predicate, action) -> { + if (predicate.test(x)) { + action.accept(x); + } + }); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + + /** Implementation of Builder. + * + * @param <G> Type of state that is created when we start processing + * @param <F> Type of state that is created when we start processing a file */ + private static class BuilderImpl<G, F> implements Builder<G, F> { + private final Supplier<G> globalStateFactory; + private final Function<G, F> fileStateFactory; + final PairList<Predicate<Line<G, F>>, Consumer<Line<G, F>>> onLineList; + final List<Consumer<Context<G, F>>> afterList; + + private BuilderImpl(Supplier<G> globalStateFactory, + Function<G, F> fileStateFactory, + PairList<Predicate<Line<G, F>>, Consumer<Line<G, F>>> onLineList, + List<Consumer<Context<G, F>>> afterList) { + this.globalStateFactory = globalStateFactory; + this.fileStateFactory = fileStateFactory; + this.onLineList = onLineList; + this.afterList = afterList; + } + + @Override public Builder<G, F> add(Predicate<Line<G, F>> linePredicate, + Consumer<Line<G, F>> action) { + onLineList.add(linePredicate, action); + return this; + } + + @Override public Builder<G, F> after(Consumer<Context<G, F>> action) { + afterList.add(action); + return this; + } + + @Override public Program<G> build() { + return new ProgramImpl<>(globalStateFactory, fileStateFactory, + onLineList.immutable(), ImmutableList.copyOf(afterList)); + } + } +} diff --git a/core/src/main/java/org/apache/calcite/util/Sources.java b/core/src/main/java/org/apache/calcite/util/Sources.java index 86cd859ad1..205a0fdfd9 100644 --- a/core/src/main/java/org/apache/calcite/util/Sources.java +++ b/core/src/main/java/org/apache/calcite/util/Sources.java @@ -23,7 +23,6 @@ import com.google.common.io.CharSource; import org.checkerframework.checker.nullness.qual.Nullable; import java.io.File; -import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; @@ -33,6 +32,7 @@ import java.net.URI; import java.net.URISyntaxException; import java.net.URL; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Paths; import java.util.Locale; import java.util.Objects; @@ -62,12 +62,18 @@ public abstract class Sources { } } - /** - * Create {@link Source} from a generic text source such as string, {@link java.nio.CharBuffer} - * or text file. Useful when data is already in memory or can't be directly read from + /** Creates a {@link Source} from a character sequence such as a + * {@link String}. */ + public static Source of(CharSequence s) { + return fromCharSource(CharSource.wrap(s)); + } + + /** Creates a {@link Source} from a generic text source such as string, + * {@link java.nio.CharBuffer} or text file. Useful when data is already + * in memory or can't be directly read from * a file or url. * - * @param source generic "re-redable" source of characters + * @param source generic "re-readable" source of characters * @return {@code Source} delegate for {@code CharSource} (can't be null) * @throws NullPointerException when {@code source} is null */ @@ -276,7 +282,7 @@ public abstract class Sources { @Override public InputStream openStream() throws IOException { if (file != null) { - return new FileInputStream(file); + return Files.newInputStream(file.toPath()); } else { return url.openStream(); } diff --git a/core/src/test/java/org/apache/calcite/test/PuffinTest.java b/core/src/test/java/org/apache/calcite/test/PuffinTest.java new file mode 100644 index 0000000000..604685e0eb --- /dev/null +++ b/core/src/test/java/org/apache/calcite/test/PuffinTest.java @@ -0,0 +1,108 @@ +/* + * 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.calcite.test; + +import org.apache.calcite.runtime.Unit; +import org.apache.calcite.util.Puffin; +import org.apache.calcite.util.Source; +import org.apache.calcite.util.Sources; + +import org.hamcrest.Matcher; +import org.junit.jupiter.api.Test; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; + +import static org.apache.calcite.test.Matchers.isLinux; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.CoreMatchers.notNullValue; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasToString; + +/** Tests {@link Puffin}. */ +public class PuffinTest { + private static final Fixture<Unit> EMPTY_FIXTURE = + new Fixture<>(Sources.of(""), Puffin.builder().build()); + + @Test void testPuffin() { + Puffin.Program<Unit> program = + Puffin.builder(() -> Unit.INSTANCE, u -> new AtomicInteger()) + .add(line -> !line.startsWith("#") + && !line.matches(".*/\\*.*\\*/.*"), + line -> line.state().incrementAndGet()) + .after(context -> + context.println("counter: " + context.state().get())) + .build(); + fixture().withDefaultInput() + .withProgram(program) + .generatesOutput(isLinux("counter: 2\n")); + } + + @Test void testEmptyProgram() { + final Puffin.Program<Unit> program = Puffin.builder().build(); + fixture().withDefaultInput() + .withProgram(program) + .generatesOutput(is("")); + } + + static Fixture<Unit> fixture() { + return EMPTY_FIXTURE; + } + + /** Fixture that contains all the state necessary to test + * {@link Puffin}. + * + * @param <G> Type of state that is created when we start processing */ + private static class Fixture<G> { + private final Source source; + private final Puffin.Program<G> program; + + Fixture(Source source, Puffin.Program<G> program) { + this.source = source; + this.program = program; + } + + public Fixture<G> withDefaultInput() { + final String inputText = "first line\n" + + "# second line\n" + + "third line /* with a comment */\n" + + "fourth line"; + return withSource(Sources.of(inputText)); + } + + private Fixture<G> withSource(Source source) { + return new Fixture<>(source, program); + } + + public <G2> Fixture<G2> withProgram(Puffin.Program<G2> program) { + return new Fixture<>(source, program); + } + + public Fixture<G> generatesOutput(Matcher<String> matcher) { + StringWriter sw = new StringWriter(); + try (PrintWriter pw = new PrintWriter(sw)) { + G g = program.execute(Stream.of(source), pw); + assertThat(g, notNullValue()); + } + assertThat(sw, hasToString(matcher)); + return this; + } + } +}
