This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit cb819196ef61882baa629eb3b93a15655aca59a3 Author: Hanyu (Peter) Zheng <[email protected]> AuthorDate: Tue Oct 27 11:20:08 2026 -0500 [FLINK-41217][table] Add ValuesSource, a FLIP-27 source over the rows of VALUES VALUES runs as an InputFormatSourceFunction, which has no state. A streaming job restored from a checkpoint taken before VALUES finished emits all its rows again. Add ValuesSource, a DataGeneratorSource over the row numbers of a VALUES clause. It checkpoints the number of the next row to emit, so a restored reader emits only the rows it has not emitted yet. ValuesCodeGenerator generates the GeneratorFunction of ValuesSource. Its map() builds the row with the given number, like the generated InputFormat builds its next row, so the rows are not kept in memory. DataGeneratorSource is always bounded. ValuesSource is unbounded, like VALUES in streaming mode. The planner doesn't use ValuesSource yet. --- flink-table/flink-table-planner/pom.xml | 7 + .../planner/codegen/ValuesCodeGenerator.scala | 78 +++++++++- .../planner/codegen/ValuesCodeGeneratorTest.java | 173 +++++++++++++++++++++ .../operators/values/ValuesGeneratorFunction.java | 56 +++++++ .../runtime/operators/values/ValuesSource.java | 54 +++++++ 5 files changed, 360 insertions(+), 8 deletions(-) diff --git a/flink-table/flink-table-planner/pom.xml b/flink-table/flink-table-planner/pom.xml index 3cd229c739f..d0694f0e431 100644 --- a/flink-table/flink-table-planner/pom.xml +++ b/flink-table/flink-table-planner/pom.xml @@ -275,6 +275,13 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-utils</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>net.javacrumbs.json-unit</groupId> <artifactId>json-unit-assertj</artifactId> diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ValuesCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ValuesCodeGenerator.scala index 2ba7da454c3..fd8589f3958 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ValuesCodeGenerator.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ValuesCodeGenerator.scala @@ -18,7 +18,11 @@ package org.apache.flink.table.planner.codegen import org.apache.flink.configuration.ReadableConfig +import org.apache.flink.connector.datagen.source.GeneratorFunction import org.apache.flink.table.data.{GenericRowData, RowData} +import org.apache.flink.table.planner.codegen.CodeGenUtils.newName +import org.apache.flink.table.planner.codegen.Indenter.toISC +import org.apache.flink.table.runtime.generated.GeneratedFunction import org.apache.flink.table.runtime.operators.values.ValuesInputFormat import org.apache.flink.table.runtime.typeutils.InternalTypeInfo import org.apache.flink.table.types.logical.RowType @@ -38,15 +42,8 @@ object ValuesCodeGenerator { tuples: util.List[util.List[RexLiteral]], description: String): ValuesInputFormat = { val ctx = new CodeGeneratorContext(tableConfig, classLoader) - val exprGenerator = new ExprCodeGenerator(ctx, false) // generate code for every record - val generatedRecords = tuples.map { - r => - exprGenerator.generateResultExpression( - r.map(exprGenerator.generateExpression), - outputType, - classOf[GenericRowData]) - } + val generatedRecords = generateRecords(ctx, outputType, tuples, reusedOutRow = true) // generate input format val generatedFunction = InputFormatCodeGenerator.generateValuesInputFormat[RowData]( @@ -58,4 +55,69 @@ object ValuesCodeGenerator { new ValuesInputFormat(generatedFunction, InternalTypeInfo.of(outputType)) } + /** + * Generates a [[GeneratorFunction]] that returns the record with the given row number. It creates + * a new record on every call. + */ + def generatorFunction( + tableConfig: ReadableConfig, + classLoader: ClassLoader, + outputType: RowType, + tuples: util.List[util.List[RexLiteral]], + description: String): GeneratedFunction[GeneratorFunction[java.lang.Long, RowData]] = { + val ctx = new CodeGeneratorContext(tableConfig, classLoader) + val generatedRecords = generateRecords(ctx, outputType, tuples, reusedOutRow = false) + val funcName = newName(ctx, description) + + val funcCode = j""" + public class $funcName implements ${classOf[GeneratorFunction[_, _]].getCanonicalName} { + + ${ctx.reuseMemberCode()} + + public $funcName(Object[] references) throws Exception { + ${ctx.reuseInitCode()} + } + + @Override + public Object map(Object rowNumber) throws Exception { + final long index = ((Long) rowNumber).longValue(); + ${generatedRecords.zipWithIndex + .map { + case (r, i) => + s""" + |if (index == ${i}L) { + | ${r.code} + | return ${r.resultTerm}; + |} + |""".stripMargin + } + .mkString("")} + throw new IllegalArgumentException( + "Row " + index + " requested, but the VALUES clause has ${tuples.size()} rows."); + } + } + """.stripMargin + + new GeneratedFunction[GeneratorFunction[java.lang.Long, RowData]]( + funcName, + funcCode, + ctx.references.toArray, + ctx.tableConfig) + } + + private def generateRecords( + ctx: CodeGeneratorContext, + outputType: RowType, + tuples: util.List[util.List[RexLiteral]], + reusedOutRow: Boolean): Seq[GeneratedExpression] = { + val exprGenerator = new ExprCodeGenerator(ctx, false) + tuples.map { + r => + exprGenerator.generateResultExpression( + r.map(exprGenerator.generateExpression), + outputType, + classOf[GenericRowData], + reusedOutRow = reusedOutRow) + } + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ValuesCodeGeneratorTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ValuesCodeGeneratorTest.java new file mode 100644 index 00000000000..670f80159ca --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/codegen/ValuesCodeGeneratorTest.java @@ -0,0 +1,173 @@ +/* + * 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.planner.codegen; + +import org.apache.flink.api.connector.source.ReaderInfo; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.lib.NumberSequenceSource.NumberSequenceSplit; +import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext; +import org.apache.flink.connector.testutils.source.reader.TestingReaderContext; +import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.planner.calcite.FlinkTypeFactory; +import org.apache.flink.table.planner.calcite.FlinkTypeSystem; +import org.apache.flink.table.runtime.operators.values.ValuesSource; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; + +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for the {@link ValuesSource} that runs the code of {@link ValuesCodeGenerator}. */ +class ValuesCodeGeneratorTest { + + private static final FlinkTypeFactory TYPE_FACTORY = + new FlinkTypeFactory( + Thread.currentThread().getContextClassLoader(), FlinkTypeSystem.INSTANCE); + + private static final RowType ROW_TYPE = RowType.of(new IntType(false), new CharType(false, 1)); + + @Test + void testEmitsEachRow() throws Exception { + ValuesSource source = source(); + assertThat(read(source, assignedSplits(source), Integer.MAX_VALUE).rows) + .containsExactly("1,a", "2,b", "3,c"); + } + + @Test + void testRestoreEmitsOnlyTheRemainingRows() throws Exception { + ValuesSource source = source(); + Read first = read(source, assignedSplits(source), 1); + assertThat(first.rows).containsExactly("1,a"); + + Read second = + read(source, serializeAndDeserialize(source, first.snapshot), Integer.MAX_VALUE); + assertThat(second.rows).containsExactly("2,b", "3,c"); + assertThat(second.snapshot).isEmpty(); + } + + /** A source for {@code VALUES (1, 'a'), (2, 'b'), (3, 'c')}. */ + private static ValuesSource source() { + RexBuilder rexBuilder = new RexBuilder(TYPE_FACTORY); + List<List<RexLiteral>> tuples = new ArrayList<>(); + for (String[] row : new String[][] {{"1", "a"}, {"2", "b"}, {"3", "c"}}) { + tuples.add( + Arrays.asList( + rexBuilder.makeExactLiteral( + new BigDecimal(row[0]), + TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER)), + rexBuilder.makeLiteral(row[1]))); + } + return new ValuesSource( + ValuesCodeGenerator.generatorFunction( + TableConfig.getDefault(), + Thread.currentThread().getContextClassLoader(), + ROW_TYPE, + tuples, + "TestValues"), + InternalTypeInfo.of(ROW_TYPE), + tuples.size()); + } + + /** The splits that the enumerator assigns to the single reader. */ + private static List<NumberSequenceSplit> assignedSplits(ValuesSource source) throws Exception { + try (MockSplitEnumeratorContext<NumberSequenceSplit> context = + new MockSplitEnumeratorContext<>(1); + SplitEnumerator<NumberSequenceSplit, Collection<NumberSequenceSplit>> enumerator = + source.createEnumerator(context)) { + enumerator.start(); + context.registerReader(new ReaderInfo(0, "localhost")); + enumerator.addReader(0); + enumerator.handleSplitRequest(0, "localhost"); + enumerator.handleSplitRequest(0, "localhost"); + assertThat(context.hasNoMoreSplits(0)).isTrue(); + return context.getSplitsAssignmentSequence().stream() + .flatMap(assignment -> assignment.assignment().get(0).stream()) + .collect(Collectors.toList()); + } + } + + /** The reader state as it comes back from a checkpoint. */ + private static List<NumberSequenceSplit> serializeAndDeserialize( + ValuesSource source, List<NumberSequenceSplit> splits) throws Exception { + SimpleVersionedSerializer<NumberSequenceSplit> serializer = source.getSplitSerializer(); + List<NumberSequenceSplit> restored = new ArrayList<>(); + for (NumberSequenceSplit split : splits) { + restored.add( + serializer.deserialize(serializer.getVersion(), serializer.serialize(split))); + } + return restored; + } + + /** Reads at most {@code maxRows} rows from the splits, then takes a snapshot. */ + private static Read read(ValuesSource source, List<NumberSequenceSplit> splits, int maxRows) + throws Exception { + TestingReaderOutput<RowData> output = new TestingReaderOutput<>(); + List<NumberSequenceSplit> snapshot; + try (SourceReader<RowData, NumberSequenceSplit> reader = + source.createReader(new TestingReaderContext())) { + reader.start(); + reader.addSplits(splits); + reader.notifyNoMoreSplits(); + InputStatus status = InputStatus.MORE_AVAILABLE; + while (output.getEmittedRecords().size() < maxRows + && status != InputStatus.END_OF_INPUT) { + status = reader.pollNext(output); + if (status == InputStatus.NOTHING_AVAILABLE) { + reader.isAvailable().get(); + } + } + snapshot = reader.snapshotState(1L); + } + // The emitted rows are read after all of them were emitted, so a row object that the + // generated code reused for the next row would show up with the values of the last row. + List<String> rows = + output.getEmittedRecords().stream() + .map(row -> row.getInt(0) + "," + row.getString(1)) + .collect(Collectors.toList()); + return new Read(rows, snapshot); + } + + private static final class Read { + final List<String> rows; + final List<NumberSequenceSplit> snapshot; + + Read(List<String> rows, List<NumberSequenceSplit> snapshot) { + this.rows = rows; + this.snapshot = snapshot; + } + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesGeneratorFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesGeneratorFunction.java new file mode 100644 index 00000000000..563a5fd553a --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesGeneratorFunction.java @@ -0,0 +1,56 @@ +/* + * 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.runtime.operators.values; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.connector.datagen.source.GeneratorFunction; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedFunction; + +/** + * Returns the row of a {@code VALUES} clause for a row number. It instantiates the generated + * function when it opens. + */ +@Internal +public class ValuesGeneratorFunction implements GeneratorFunction<Long, RowData> { + + private static final long serialVersionUID = 1L; + + private final GeneratedFunction<GeneratorFunction<Long, RowData>> generatedFunction; + + private transient GeneratorFunction<Long, RowData> function; + + public ValuesGeneratorFunction( + GeneratedFunction<GeneratorFunction<Long, RowData>> generatedFunction) { + this.generatedFunction = generatedFunction; + } + + @Override + public void open(SourceReaderContext readerContext) throws Exception { + function = + generatedFunction.newInstance( + readerContext.getUserCodeClassLoader().asClassLoader()); + } + + @Override + public RowData map(Long rowNumber) throws Exception { + return function.map(rowNumber); + } +} diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesSource.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesSource.java new file mode 100644 index 00000000000..d3be639f85f --- /dev/null +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/values/ValuesSource.java @@ -0,0 +1,54 @@ +/* + * 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.runtime.operators.values; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.connector.datagen.source.DataGeneratorSource; +import org.apache.flink.connector.datagen.source.GeneratorFunction; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; + +/** + * A {@link DataGeneratorSource} over the rows of a {@code VALUES} clause in streaming mode. It + * checkpoints the number of the next row to emit, so a restored job emits only the rows it has not + * emitted yet. + * + * <p>Unlike {@link DataGeneratorSource}, it is unbounded, so that a pipeline in {@code AUTOMATIC} + * runtime mode, e.g. one that is converted to a {@code DataStream}, keeps running in streaming + * mode. + */ +@Internal +public class ValuesSource extends DataGeneratorSource<RowData> { + + private static final long serialVersionUID = 1L; + + public ValuesSource( + GeneratedFunction<GeneratorFunction<Long, RowData>> generatedFunction, + InternalTypeInfo<RowData> producedType, + int numRows) { + super(new ValuesGeneratorFunction(generatedFunction), numRows, producedType); + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.CONTINUOUS_UNBOUNDED; + } +}
