This is an automated email from the ASF dual-hosted git repository. godfrey pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2176ec0aa131cd7dbbf4f19ac02640084b6db021 Author: godfreyhe <[email protected]> AuthorDate: Thu Aug 4 16:20:51 2022 +0800 [FLINK-28710][table-planner] The fact source of DPP must be FLIP-27 source This closes #20472 --- .../planner/connectors/DynamicSourceUtils.java | 20 +++ .../batch/DynamicPartitionPruningRule.java | 31 ++-- .../flink/connector/source/ValuesSource.java | 120 +++++++++++++ .../flink/connector/source/ValuesSourceReader.java | 147 ++++++++++++++++ .../connector/source/enumerator/NoOpEnumState.java | 22 +++ .../source/enumerator/NoOpEnumStateSerializer.java | 41 +++++ .../source/enumerator/ValuesSourceEnumerator.java | 81 +++++++++ .../connector/source/split/ValuesSourceSplit.java | 41 +++++ .../source/split/ValuesSourceSplitSerializer.java | 55 ++++++ .../planner/factories/TestValuesTableFactory.java | 12 +- .../batch/DynamicPartitionPruningRuleTest.java | 193 ++++++++++++--------- .../batch/DynamicPartitionPruningRuleTest.xml | 24 +++ 12 files changed, 687 insertions(+), 100 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java index 989a20c6670..c2497cfbb36 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSourceUtils.java @@ -19,8 +19,10 @@ package org.apache.flink.table.planner.connectors; import org.apache.flink.annotation.Internal; +import org.apache.flink.api.dag.Transformation; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.transformations.SourceTransformation; import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.ValidationException; @@ -36,6 +38,7 @@ import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider; +import org.apache.flink.table.connector.source.SourceProvider; import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata; import org.apache.flink.table.planner.calcite.FlinkRelBuilder; import org.apache.flink.table.planner.expressions.converter.ExpressionConverter; @@ -62,6 +65,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -255,6 +259,22 @@ public final class DynamicSourceUtils { return isCDCSource && changeEventsDuplicate && hasPrimaryKey; } + /** Returns true if the source is FLIP-27 source, else false. */ + public static boolean isNewSource(ScanTableSource scanTableSource) { + ScanTableSource.ScanRuntimeProvider provider = + scanTableSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE); + if (provider instanceof SourceProvider) { + return true; + } else if (provider instanceof TransformationScanProvider) { + Transformation<?> transformation = + ((TransformationScanProvider) provider) + .createTransformation(name -> Optional.empty()); + return transformation instanceof SourceTransformation; + } + // TODO supports more + return false; + } + // -------------------------------------------------------------------------------------------- /** Creates a specialized node for assigning watermarks. */ diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java index 2a5ce1f1288..79317c1f826 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java @@ -22,8 +22,10 @@ import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.config.OptimizerConfigOptions; import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; +import org.apache.flink.table.planner.connectors.DynamicSourceUtils; import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec; import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec; import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc; @@ -75,7 +77,7 @@ import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPT * * <p>Suppose we have the original physical plan: * - * <pre>{@Code + * <pre>{@code * LogicalProject(...) * HashJoin(joinType=[InnerJoin], where=[=(fact_partition_key, dim_key)], select=[xxx]) * * :- TableSourceScan(table=[[fact]], fields=[xxx, fact_partition_key],) # Is a partition table. @@ -86,7 +88,7 @@ import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPT * * <p>This physical plan will be rewritten to: * - * <pre>{@Code + * <pre>{@code * HashJoin(joinType=[InnerJoin], where=[=(fact_partition_key, dim_key)], select=[xxx]) * :- DynamicFilteringTableSourceScan(table=[[fact]], fields=[xxx, fact_partition_key]) # Is a partition table. * : +- DynamicFilteringDataCollector(fields=[dim_key]) @@ -170,7 +172,12 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config return false; } DynamicTableSource tableSource = tableSourceTable.tableSource(); - if (!(tableSource instanceof SupportsDynamicFiltering)) { + if (!(tableSource instanceof SupportsDynamicFiltering) + || !(tableSource instanceof ScanTableSource)) { + return false; + } + + if (!DynamicSourceUtils.isNewSource((ScanTableSource) tableSource)) { return false; } @@ -196,17 +203,10 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config } else { // Changing the fact key index in fact table calc output to fact key index in fact // table, and filtering these fields that computing in calc node. - RexProgram origProgram = factCalc.getProgram(); + RexProgram program = factCalc.getProgram(); List<Integer> joinKeysIndexInFactTable = new ArrayList<>(); - List<RexNode> projectInJoinKeyList = - factJoinKeys.stream() - .map( - i -> - origProgram.expandLocalRef( - origProgram.getProjectList().get(i))) - .collect(Collectors.toList()); - - for (RexNode node : projectInJoinKeyList) { + for (int joinKeyIdx : factJoinKeys) { + RexNode node = program.expandLocalRef(program.getProjectList().get(joinKeyIdx)); if (node instanceof RexInputRef) { joinKeysIndexInFactTable.add(((RexInputRef) node).getIndex()); } @@ -229,8 +229,9 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config if (!candidateFields.contains(field)) { throw new TableException( String.format( - "Field %s not in join key, please verify the applyDynamicFiltering method In %s", - field, tableSource.asSummaryString())); + "Field: %s does not exist in the given fields: %s, " + + "please verify the applyDynamicFiltering method in connector: %s", + field, candidateFields, tableSource.asSummaryString())); } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSource.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSource.java new file mode 100644 index 00000000000..6d9a3e186e2 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSource.java @@ -0,0 +1,120 @@ +/* + * 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.connector.source; + +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.Source; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.source.enumerator.NoOpEnumState; +import org.apache.flink.connector.source.enumerator.NoOpEnumStateSerializer; +import org.apache.flink.connector.source.enumerator.ValuesSourceEnumerator; +import org.apache.flink.connector.source.split.ValuesSourceSplit; +import org.apache.flink.connector.source.split.ValuesSourceSplitSerializer; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.data.RowData; +import org.apache.flink.util.Preconditions; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +/** + * A {@link Source} implementation that reads data from a list. + * + * <p>The source is useful for FLIP-27 source tests. + * + * <p>{@code FromElementsSource} requires the elements must be serializable, and the parallelism + * must be 1. RowData is not serializable and the parallelism of table source may not be 1, so we + * introduce a new source for testing in table module. + */ +public class ValuesSource implements Source<RowData, ValuesSourceSplit, NoOpEnumState> { + private final TypeSerializer<RowData> serializer; + + private final List<byte[]> serializedElements; + + public ValuesSource(Collection<RowData> elements, TypeSerializer<RowData> serializer) { + Preconditions.checkState(serializer != null, "serializer not set"); + this.serializedElements = serializeElements(elements, serializer); + this.serializer = serializer; + } + + private List<byte[]> serializeElements( + Collection<RowData> elements, TypeSerializer<RowData> serializer) { + List<byte[]> serializeElements = new ArrayList<>(); + + for (RowData element : elements) { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos)) { + serializer.serialize(element, wrapper); + serializeElements.add(baos.toByteArray()); + } catch (Exception e) { + throw new TableException( + "Serializing the source elements failed: " + e.getMessage(), e); + } + } + return serializeElements; + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public SourceReader<RowData, ValuesSourceSplit> createReader(SourceReaderContext readerContext) + throws Exception { + return new ValuesSourceReader(serializedElements, serializer, readerContext); + } + + @Override + public SplitEnumerator<ValuesSourceSplit, NoOpEnumState> createEnumerator( + SplitEnumeratorContext<ValuesSourceSplit> enumContext) { + List<ValuesSourceSplit> splits = + IntStream.range(0, serializedElements.size()) + .mapToObj(ValuesSourceSplit::new) + .collect(Collectors.toList()); + return new ValuesSourceEnumerator(enumContext, splits); + } + + @Override + public SplitEnumerator<ValuesSourceSplit, NoOpEnumState> restoreEnumerator( + SplitEnumeratorContext<ValuesSourceSplit> enumContext, NoOpEnumState checkpoint) + throws Exception { + throw new UnsupportedOperationException("Unsupported now."); + } + + @Override + public SimpleVersionedSerializer<ValuesSourceSplit> getSplitSerializer() { + return new ValuesSourceSplitSerializer(); + } + + @Override + public SimpleVersionedSerializer<NoOpEnumState> getEnumeratorCheckpointSerializer() { + return new NoOpEnumStateSerializer(); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSourceReader.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSourceReader.java new file mode 100644 index 00000000000..84e33e529da --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/ValuesSourceReader.java @@ -0,0 +1,147 @@ +/* + * 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.connector.source; + +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.connector.source.ReaderOutput; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.connector.source.split.ValuesSourceSplit; +import org.apache.flink.core.io.InputStatus; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.data.RowData; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayInputStream; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; + +/** A {@link SourceReader} implementation that reads data from a list. */ +public class ValuesSourceReader implements SourceReader<RowData, ValuesSourceSplit> { + private static final Logger LOG = LoggerFactory.getLogger(ValuesSourceReader.class); + + /** The context for this reader, to communicate with the enumerator. */ + private final SourceReaderContext context; + + /** The availability future. This reader is available as soon as a split is assigned. */ + private CompletableFuture<Void> availability; + + private final List<byte[]> serializedElements; + private final TypeSerializer<RowData> serializer; + private List<RowData> elements; + + /** The remaining splits that were assigned but not yet processed. */ + private final Queue<ValuesSourceSplit> remainingSplits; + + private boolean noMoreSplits; + + public ValuesSourceReader( + List<byte[]> serializedElements, + TypeSerializer<RowData> serializer, + SourceReaderContext context) { + this.serializedElements = serializedElements; + this.serializer = serializer; + this.context = context; + this.availability = new CompletableFuture<>(); + this.remainingSplits = new ArrayDeque<>(); + } + + @Override + public void start() { + elements = new ArrayList<>(); + for (byte[] bytes : serializedElements) { + try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes)) { + DataInputView input = new DataInputViewStreamWrapper(bais); + RowData element = serializer.deserialize(input); + elements.add(element); + } catch (Exception e) { + throw new TableException( + "Failed to deserialize an element from the source. " + + "If you are using user-defined serialization (Value and Writable types), check the " + + "serialization functions.\nSerializer is " + + serializer, + e); + } + } + // request a split if we don't have one + if (remainingSplits.isEmpty()) { + context.sendSplitRequest(); + } + } + + @Override + public InputStatus pollNext(ReaderOutput<RowData> output) throws Exception { + ValuesSourceSplit currentSplit = remainingSplits.poll(); + if (currentSplit != null) { + output.collect(elements.get(currentSplit.getIndex())); + // request another split + context.sendSplitRequest(); + return InputStatus.MORE_AVAILABLE; + } else if (noMoreSplits) { + return InputStatus.END_OF_INPUT; + } else { + // ensure we are not called in a loop by resetting the availability future + if (availability.isDone()) { + availability = new CompletableFuture<>(); + } + + return InputStatus.NOTHING_AVAILABLE; + } + } + + @Override + public List<ValuesSourceSplit> snapshotState(long checkpointId) { + return Collections.emptyList(); + } + + @Override + public CompletableFuture<Void> isAvailable() { + return availability; + } + + @Override + public void addSplits(List<ValuesSourceSplit> splits) { + remainingSplits.addAll(splits); + // set availability so that pollNext is actually called + availability.complete(null); + } + + @Override + public void notifyNoMoreSplits() { + noMoreSplits = true; + // set availability so that pollNext is actually called + availability.complete(null); + } + + @Override + public void close() throws Exception {} + + @Override + public void notifyCheckpointComplete(long checkpointId) throws Exception { + LOG.info("checkpoint {} finished.", checkpointId); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumState.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumState.java new file mode 100644 index 00000000000..bd91a0ecfb7 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumState.java @@ -0,0 +1,22 @@ +/* + * 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.connector.source.enumerator; + +/** Mock enumerator state. */ +public class NoOpEnumState {} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumStateSerializer.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumStateSerializer.java new file mode 100644 index 00000000000..60df9d04c93 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/NoOpEnumStateSerializer.java @@ -0,0 +1,41 @@ +/* + * 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.connector.source.enumerator; + +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.io.IOException; + +/** Mock enumerator state serializer. */ +public class NoOpEnumStateSerializer implements SimpleVersionedSerializer<NoOpEnumState> { + @Override + public int getVersion() { + return 0; + } + + @Override + public byte[] serialize(NoOpEnumState obj) throws IOException { + return new byte[0]; + } + + @Override + public NoOpEnumState deserialize(int version, byte[] serialized) throws IOException { + return new NoOpEnumState(); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/ValuesSourceEnumerator.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/ValuesSourceEnumerator.java new file mode 100644 index 00000000000..4e6f2d287c8 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/enumerator/ValuesSourceEnumerator.java @@ -0,0 +1,81 @@ +/* + * 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.connector.source.enumerator; + +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.source.ValuesSource; +import org.apache.flink.connector.source.split.ValuesSourceSplit; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.List; +import java.util.Queue; + +/** + * A {@link SplitEnumerator} for {@link ValuesSource}. Simply takes the pre-split set of splits and + * assigns it first-come-first-serve. + */ +public class ValuesSourceEnumerator implements SplitEnumerator<ValuesSourceSplit, NoOpEnumState> { + + private final SplitEnumeratorContext<ValuesSourceSplit> context; + private final Queue<ValuesSourceSplit> remainingSplits; + + public ValuesSourceEnumerator( + SplitEnumeratorContext<ValuesSourceSplit> context, + List<ValuesSourceSplit> remainingSplits) { + this.context = context; + this.remainingSplits = new ArrayDeque<>(remainingSplits); + } + + @Override + public void start() {} + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + final ValuesSourceSplit nextSplit = remainingSplits.poll(); + if (nextSplit != null) { + context.assignSplit(nextSplit, subtaskId); + } else { + context.signalNoMoreSplits(subtaskId); + } + } + + @Override + public void addSplitsBack(List<ValuesSourceSplit> splits, int subtaskId) { + remainingSplits.addAll(splits); + } + + @Override + public void addReader(int subtaskId) { + // we don't assign any splits here, because this registration happens after fist startup and + // after each reader restart/recovery we only want to assign splits once, initially, which + // we get by reacting to the readers explicit split request + } + + @Override + public NoOpEnumState snapshotState(long checkpointId) throws Exception { + return new NoOpEnumState(); + } + + @Override + public void close() throws IOException {} +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplit.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplit.java new file mode 100644 index 00000000000..87d4b9bce9e --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplit.java @@ -0,0 +1,41 @@ +/* + * 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.connector.source.split; + +import org.apache.flink.api.connector.source.SourceSplit; +import org.apache.flink.connector.source.ValuesSource; + +/** The split of the {@link ValuesSource}. */ +public class ValuesSourceSplit implements SourceSplit { + + private final int index; + + public ValuesSourceSplit(int index) { + this.index = index; + } + + public int getIndex() { + return index; + } + + @Override + public String splitId() { + return "" + index; + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplitSerializer.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplitSerializer.java new file mode 100644 index 00000000000..697162e18b4 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/source/split/ValuesSourceSplitSerializer.java @@ -0,0 +1,55 @@ +/* + * 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.connector.source.split; + +import org.apache.flink.connector.source.ValuesSource; +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +/** The split serializer for the {@link ValuesSource}. */ +public class ValuesSourceSplitSerializer implements SimpleVersionedSerializer<ValuesSourceSplit> { + @Override + public int getVersion() { + return 0; + } + + @Override + public byte[] serialize(ValuesSourceSplit split) throws IOException { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(baos)) { + out.writeInt(split.getIndex()); + out.flush(); + return baos.toByteArray(); + } + } + + @Override + public ValuesSourceSplit deserialize(int version, byte[] serialized) throws IOException { + try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized); + DataInputStream in = new DataInputStream(bais)) { + int index = in.readInt(); + return new ValuesSourceSplit(index); + } + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java index 42dafdd58c6..7960f071e50 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java @@ -27,6 +27,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.io.CollectionInputFormat; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.connector.source.ValuesSource; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSink; import org.apache.flink.streaming.api.datastream.DataStreamSource; @@ -55,6 +56,7 @@ import org.apache.flink.table.connector.source.InputFormatProvider; import org.apache.flink.table.connector.source.LookupTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceFunctionProvider; +import org.apache.flink.table.connector.source.SourceProvider; import org.apache.flink.table.connector.source.TableFunctionProvider; import org.apache.flink.table.connector.source.abilities.SupportsAggregatePushDown; import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering; @@ -267,7 +269,9 @@ public final class TestValuesTableFactory private static final ConfigOption<String> RUNTIME_SOURCE = ConfigOptions.key("runtime-source") .stringType() - .defaultValue("SourceFunction"); // another is "InputFormat" + .defaultValue("SourceFunction") + .withDescription( + "Accepted values are: SourceFunction, InputFormat, DataStream, NewSource"); private static final ConfigOption<Boolean> FAILING_SOURCE = ConfigOptions.key("failing-source").booleanType().defaultValue(false); @@ -534,7 +538,7 @@ public final class TestValuesTableFactory helper.getOptions().get(WRITABLE_METADATA), context.getClassLoader()); final ChangelogMode changelogMode = Optional.ofNullable(helper.getOptions().get(SINK_CHANGELOG_MODE_ENFORCED)) - .map(m -> parseChangelogMode(m)) + .map(this::parseChangelogMode) .orElse(null); final DataType consumedType = context.getCatalogTable().getSchema().toPhysicalRowDataType(); @@ -863,6 +867,10 @@ public final class TestValuesTableFactory } catch (IOException e) { throw new TableException("Fail to init data stream source", e); } + case "NewSource": + checkArgument( + !failingSource, "Values Source doesn't support as failing new source."); + return SourceProvider.of(new ValuesSource(values, serializer)); default: throw new IllegalArgumentException( "Unsupported runtime source class: " + runtimeSource); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java index 5901b06c2e2..dec652bb98e 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java @@ -35,8 +35,9 @@ import org.junit.Test; * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}. */ public class DynamicPartitionPruningRuleTest extends TableTestBase { - protected BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault()); - TestValuesCatalog catalog = new TestValuesCatalog("testCatalog", "test_database", true); + private final BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault()); + private final TestValuesCatalog catalog = + new TestValuesCatalog("testCatalog", "test_database", true); @Before public void setup() { @@ -47,35 +48,36 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED, true); // partition fact table. - String ddl1 = - "CREATE TABLE test_database.fact_part (\n" - + " id BIGINT,\n" - + " name STRING,\n" - + " amount BIGINT,\n" - + " price BIGINT,\n" - + " fact_date_sk BIGINT\n" - + ") PARTITIONED BY (fact_date_sk)\n" - + "WITH (\n" - + " 'connector' = 'values',\n" - + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n" - + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl1); + util.tableEnv() + .executeSql( + "CREATE TABLE fact_part (\n" + + " id BIGINT,\n" + + " name STRING,\n" + + " amount BIGINT,\n" + + " price BIGINT,\n" + + " fact_date_sk BIGINT\n" + + ") PARTITIONED BY (fact_date_sk)\n" + + "WITH (\n" + + " 'connector' = 'values',\n" + + " 'runtime-source' = 'NewSource',\n" + + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n" + + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n" + + " 'bounded' = 'true'\n" + + ")"); // dim table. - String ddl2 = - "CREATE TABLE test_database.dim (\n" - + " id BIGINT,\n" - + " male BOOLEAN,\n" - + " amount BIGINT,\n" - + " price BIGINT,\n" - + " dim_date_sk BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl2); + util.tableEnv() + .executeSql( + "CREATE TABLE dim (\n" + + " id BIGINT,\n" + + " male BOOLEAN,\n" + + " amount BIGINT,\n" + + " price BIGINT,\n" + + " dim_date_sk BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"); } @Test @@ -109,19 +111,20 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { public void testFactTableIsNotPartitionTable() { // non-partition fact table. Dynamic partition pruning will not succeed if fact side is not // partition table. - String ddl1 = - "CREATE TABLE test_database.none_part_fact (\n" - + " id BIGINT,\n" - + " name STRING,\n" - + " amount BIGINT,\n" - + " price BIGINT,\n" - + " fact_date_sk BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl1); + util.tableEnv() + .executeSql( + "CREATE TABLE none_part_fact (\n" + + " id BIGINT,\n" + + " name STRING,\n" + + " amount BIGINT,\n" + + " price BIGINT,\n" + + " fact_date_sk BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'runtime-source' = 'NewSource',\n" + + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n" + + " 'bounded' = 'true'\n" + + ")"); String query = "Select * from dim, none_part_fact where none_part_fact.fact_date_sk = dim.dim_date_sk" @@ -129,6 +132,30 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { util.verifyRelPlan(query); } + @Test + public void testFactTableIsLegacySource() { + util.tableEnv() + .executeSql( + "CREATE TABLE legacy_source (\n" + + " id BIGINT,\n" + + " name STRING,\n" + + " amount BIGINT,\n" + + " price BIGINT,\n" + + " fact_date_sk BIGINT\n" + + ") PARTITIONED BY (fact_date_sk)\n" + + "WITH (\n" + + " 'connector' = 'values',\n" + + " 'runtime-source' = 'SourceFunction',\n" + + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n" + + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n" + + " 'bounded' = 'true'\n" + + ")"); + String query = + "Select * from dim, legacy_source where legacy_source.fact_date_sk = dim.dim_date_sk" + + " and dim.price < 500"; + util.verifyRelPlan(query); + } + @Test public void testDimTableWithFilterPushDown() { // Even though have filter push down, dynamic partition pruning will succeed. @@ -308,16 +335,16 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { @Test public void testMultiJoin() { // Another table. - String ddl = - "CREATE TABLE test_database.sales (\n" - + " id BIGINT,\n" - + " amount BIGINT,\n" - + " price BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl); + util.tableEnv() + .executeSql( + "CREATE TABLE sales (\n" + + " id BIGINT,\n" + + " amount BIGINT,\n" + + " price BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"); String query = "Select * from fact_part, dim, sales where fact_part.id = sales.id and" @@ -328,27 +355,27 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { @Test public void testComplexDimSideWithJoinInDimSide() { // Dim side contains join will not succeed in this version, it will improve later. - String ddl1 = - "CREATE TABLE test_database.sales (\n" - + " id BIGINT,\n" - + " amount BIGINT,\n" - + " price BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl1); - - String ddl2 = - "CREATE TABLE test_database.item (\n" - + " id BIGINT,\n" - + " amount BIGINT,\n" - + " price BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl2); + util.tableEnv() + .executeSql( + "CREATE TABLE sales (\n" + + " id BIGINT,\n" + + " amount BIGINT,\n" + + " price BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"); + + util.tableEnv() + .executeSql( + "CREATE TABLE item (\n" + + " id BIGINT,\n" + + " amount BIGINT,\n" + + " price BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"); String query = "Select * from fact_part join" @@ -361,16 +388,16 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { @Test public void testComplexDimSideWithAggInDimSide() { // Dim side contains agg will not succeed in this version, it will improve later. - String ddl = - "CREATE TABLE test_database.sales (\n" - + " id BIGINT,\n" - + " amount BIGINT,\n" - + " price BIGINT\n" - + ") WITH (\n" - + " 'connector' = 'values',\n" - + " 'bounded' = 'true'\n" - + ")"; - util.tableEnv().executeSql(ddl); + util.tableEnv() + .executeSql( + "CREATE TABLE sales (\n" + + " id BIGINT,\n" + + " amount BIGINT,\n" + + " price BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"); String query = "Select * from fact_part join" diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml index 0eb8ec031a6..1531fd0b7a8 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml @@ -478,6 +478,30 @@ NestedLoopJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk), =( +- DynamicFilteringDataCollector(fields=[dim_date_sk]) +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price, 500)]) +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) +]]> + </Resource> + </TestCase> + <TestCase name="testFactTableIsLegacySource"> + <Resource name="sql"> + <![CDATA[Select * from dim, legacy_source where legacy_source.fact_date_sk = dim.dim_date_sk and dim.price < 500]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(id=[$0], male=[$1], amount=[$2], price=[$3], dim_date_sk=[$4], id0=[$5], name=[$6], amount0=[$7], price0=[$8], fact_date_sk=[$9]) ++- LogicalFilter(condition=[AND(=($9, $4), <($3, 500))]) + +- LogicalJoin(condition=[true], joinType=[inner]) + :- LogicalTableScan(table=[[testCatalog, test_database, dim]]) + +- LogicalTableScan(table=[[testCatalog, test_database, legacy_source]]) +]]> + </Resource> + <Resource name="optimized rel plan"> + <![CDATA[ +HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)], select=[id, male, amount, price, dim_date_sk, id0, name, amount0, price0, fact_date_sk], build=[left]) +:- Exchange(distribution=[hash[dim_date_sk]]) +: +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price, 500)]) +: +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) ++- Exchange(distribution=[hash[fact_date_sk]]) + +- TableSourceScan(table=[[testCatalog, test_database, legacy_source]], fields=[id, name, amount, price, fact_date_sk]) ]]> </Resource> </TestCase>
