MartijnVisser commented on code in PR #174:
URL:
https://github.com/apache/flink-connector-kafka/pull/174#discussion_r4069772654
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java:
##########
@@ -264,7 +267,7 @@ private static final class OutputProjectionCollector
@Override
public void collect(RowData physicalValueRow) {
// no key defined
- if (keyProjection.length == 0) {
+ if (keyProjector.isEmptyProjection()) {
Review Comment:
This used to mean the table has no key, it now also means the query selects
no key column. A record with a null Kafka key is then emitted rather than
skipped, so `SELECT c` returns a row that `SELECT a, c` does not. Can it stay
on `keyDeserialization == null`?
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java:
##########
@@ -74,13 +75,15 @@
import java.util.Properties;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
Review Comment:
`mvn spotless:check` fails on this import and on the same one in
DynamicKafkaTableSource, so CI stops before any test runs. Both went unused
when the rebase took main's deletion of `getResetStrategy`.
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -1721,6 +1750,504 @@ private void
testStartFromGroupOffsetsWithNoneResetStrategy(final String format)
}
}
+ private void projectionPushdownSetupData(final String format, final String
topic)
+ throws Exception {
+ createTestTopic(topic, 1, 1);
+
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `a` STRING,\n"
+ + " `b` STRING,\n"
+ + " `topic` STRING NOT NULL METADATA
VIRTUAL,\n"
+ + " `c` STRING,\n"
+ + " `partition` INT NOT NULL METADATA
VIRTUAL,\n"
+ + " `d` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'kafka',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' =
'earliest-offset',\n"
+ + " %s,\n"
+ + " 'key.fields' = 'a; b',\n"
+ + " %s,\n"
+ + " 'value.fields-include' = 'EXCEPT_KEY'\n"
+ + ")",
+ topic,
+ bootstraps,
+ groupId,
+ keyFormatOptions(format),
+ valueFormatOptions(format));
+ tEnv.executeSql(createTable);
+
+ final String initialValues = "INSERT INTO kafka (a, b, c, d) SELECT
'a', 'b', 'c', 'd'";
+ tEnv.executeSql(initialValues).await();
+ }
+
+ @ParameterizedTest(name = "format: {0}")
+ @MethodSource("formats")
+ public void testProjectionPushdownSelectAllFields(final String format)
throws Exception {
+ final String topic = "testProjectionPushdown_" + format + "_" +
UUID.randomUUID();
+ projectionPushdownSetupData(format, topic);
+
+ assertQueryResult(
+ "SELECT * FROM kafka",
+ "== Optimized Execution Plan ==\n"
+ + "Calc(select=[a, b, topic, c, partition, d])\n"
+ + "+- TableSourceScan(table=[[default_catalog,
default_database, kafka, metadata=[topic, partition]]], fields=[a, b, c, d,
topic, partition])\n",
+ Collections.singletonList(String.format("+I(a,b,%s,c,%d,d)",
topic, 0)));
+
+ cleanupTopic(topic);
+ }
+
+ @ParameterizedTest(name = "format: {0}")
+ @MethodSource("formats")
+ public void testProjectionPushdownSelectSpecificPhysicalFields(final
String format)
Review Comment:
None of these selects only value columns, so the branch is never reached
with a key format present, the case where it differs from before. I think a
case for that, plus one for a null key, would cover it.
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/Decoder.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * 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.streaming.connectors.kafka.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.table.connector.Projection;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
+import org.apache.flink.table.connector.source.DynamicTableSource.Context;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Decoding messages consists of two potential steps:
+ *
+ * <ol>
+ * <li>Deserialization i.e deserializing the {@code byte[]} into a {@link
RowData}. This process
+ * is handled by a {@link DeserializationSchema}.
+ * <li>Projection i.e. projecting any required fields from the deserialized
{@link RowData}
+ * (returned by the {@link DeserializationSchema} in the first step) to
their positions in the
+ * final produced {@link RowData}. This process is handled by a {@link
Projector}.
+ * </ol>
+ *
+ * <p>In order to decode messages correctly, the {@link DeserializationSchema}
and the {@link
+ * Projector} need to work together. For example, the {@link Projector} needs
to know the positions
+ * of the required fields in the {@link RowData} returned by the {@link
DeserializationSchema} in
+ * order to be able to correctly set fields in the final produced {@link
RowData}.
+ *
+ * <p>That's why we have this {@link Decoder} class. This class ensures that
the returned {@link
+ * DeserializationSchema} and {@link Projector} will work together to decode
messages correctly.
+ */
+@Internal
+public class Decoder {
+
+ /**
+ * Can be null. Null is used inside {@link
DynamicKafkaDeserializationSchema} to avoid
+ * deserializing keys if not required.
+ */
+ private final @Nullable DeserializationSchema<RowData>
deserializationSchema;
+
+ /** Mapping of the physical position in the key to the target position in
the RowData. */
+ private final Projector projector;
+
+ private Decoder(
+ final DeserializationSchema<RowData> deserializationSchema, final
Projector projector) {
+ this.deserializationSchema = deserializationSchema;
+ this.projector = projector;
+ }
+
+ /**
+ * @param decodingFormat Optional format for decoding bytes.
+ * @param physicalTableDataType The data type representing the table
schema.
+ * @param physicalDataTypeProjection Indices indicate the position of the
field in the dataType
+ * (key/value). Values indicate the position of the field in the
tableSchema.
+ * @param prefix Optional field prefix
+ * @param projectedPhysicalFields Indices indicate the position of the
field in the produced
+ * Row. Values indicate the position of the field in the table schema.
+ * @param pushProjectionsIntoDecodingFormat if this is true and the format
is a {@link
+ * ProjectableDecodingFormat}, any {@param projectedPhysicalFields}
will be pushed down into
+ * the {@link ProjectableDecodingFormat}. Otherwise, projections will
be applied after
+ * deserialization.
+ * @return a {@link Decoder} instance.
+ */
+ public static Decoder create(
+ final Context context,
+ final @Nullable DecodingFormat<DeserializationSchema<RowData>>
decodingFormat,
+ final DataType physicalTableDataType,
+ final int[] physicalDataTypeProjection,
+ final @Nullable String prefix,
+ final int[][] projectedPhysicalFields,
+ final List<String> metadataKeys,
+ final boolean pushProjectionsIntoDecodingFormat) {
+ if (decodingFormat == null) {
+ return Decoder.noDeserializationOrProjection();
+ } else if (!pushProjectionsIntoDecodingFormat
+ || !(decodingFormat instanceof ProjectableDecodingFormat)) {
+ return Decoder.projectAfterDeserializing(
+ context,
+ decodingFormat,
+ physicalTableDataType,
+ physicalDataTypeProjection,
+ prefix,
+ projectedPhysicalFields,
+ metadataKeys);
+ } else {
+ final ProjectableDecodingFormat<DeserializationSchema<RowData>>
+ projectableDecodingFormat =
+
(ProjectableDecodingFormat<DeserializationSchema<RowData>>)
+ decodingFormat;
+ if (projectableDecodingFormat.supportsNestedProjection()) {
+ return Decoder.projectInsideDeserializer(
+ context,
+ projectableDecodingFormat,
+ physicalTableDataType,
+ physicalDataTypeProjection,
+ prefix,
+ projectedPhysicalFields,
+ metadataKeys);
+ } else {
+ return
Decoder.projectTopLevelInsideDeserializerThenNestedAfter(
+ context,
+ projectableDecodingFormat,
+ physicalTableDataType,
+ physicalDataTypeProjection,
+ prefix,
+ projectedPhysicalFields,
+ metadataKeys);
+ }
+ }
+ }
+
+ /**
+ * @return a {@link DeserializationSchema} or null.
+ */
+ @Nullable
+ public DeserializationSchema<RowData> getDeserializationSchema() {
+ return deserializationSchema;
+ }
+
+ /**
+ * @return a {@link Projector}.
+ */
+ public Projector getProjector() {
+ return projector;
+ }
+
+ /**
+ * Creates an identity projection array where each field in the row type
maps to itself.
+ *
+ * @param rowType the row type representing the table schema
+ * @return an int[][] with one entry per field in the row type, where
entry i is {i}
+ */
+ public static int[][] identityProjection(final RowType rowType) {
+ final int tableSchemaSize = rowType.getFieldCount();
+ final int[][] projectedFields = new int[tableSchemaSize][];
+ for (int i = 0; i < tableSchemaSize; i++) {
+ projectedFields[i] = new int[] {i};
+ }
+ return projectedFields;
+ }
+
+ private static Decoder noDeserializationOrProjection() {
+ return new Decoder(null, new ProjectorImpl(Collections.emptyMap(), 0,
0));
+ }
+
+ private static DataType toPhysicalDataType(
+ final DataType physicalTableDataType,
+ final int[] physicalDataTypeProjection,
+ final @Nullable String prefix) {
+ final DataType temp =
+
Projection.of(physicalDataTypeProjection).project(physicalTableDataType);
+ return Optional.ofNullable(prefix)
+ .map(s -> TableDataTypeUtils.stripRowPrefix(temp, s))
+ .orElse(temp);
+ }
+
+ private static Map<Integer, Integer> tableToDeserializedTopLevelPos(
+ final int[] dataTypeProjection) {
+ final HashMap<Integer, Integer> tableToDeserializedPos = new
HashMap<>();
+ for (int i = 0; i < dataTypeProjection.length; i++) {
+ tableToDeserializedPos.put(dataTypeProjection[i], i);
+ }
+ return tableToDeserializedPos;
+ }
+
+ private static int[] copyArray(final int[] arr) {
+ return Arrays.copyOf(arr, arr.length);
+ }
+
+ private static void addMetadataProjections(
+ final DecodingFormat<?> decodingFormat,
+ final int deserializedSize,
+ final int physicalSize,
+ final List<String> requestedMetadataKeys,
+ final Map<List<Integer>, Integer> deserializedToProducedPos) {
+
+ if (!requestedMetadataKeys.isEmpty()) {
+ decodingFormat.applyReadableMetadata(requestedMetadataKeys);
+
+ // project only requested metadata keys
+ for (int i = 0; i < requestedMetadataKeys.size(); i++) {
+ // metadata is always added to the end of the deserialized row
by the DecodingFormat
+ final int deserializedPos = deserializedSize + i;
+ // we need to always add metadata to the end of the produced
row
+ final int producePos = physicalSize + i;
+ deserializedToProducedPos.put(
+ Collections.singletonList(deserializedPos),
producePos);
+ }
+ }
+ }
+
+ /**
+ * This method generates a {@link Decoder} which pushes projections down
directly into the
+ * {@link ProjectableDecodingFormat} which takes care of projecting the
fields during the
+ * deserialization process itself.
+ */
+ private static Decoder projectInsideDeserializer(
+ final Context context,
+ final ProjectableDecodingFormat<DeserializationSchema<RowData>>
+ projectableDecodingFormat,
+ final DataType physicalTableDataType,
+ final int[] physicalDataTypeProjection,
+ final @Nullable String prefix,
+ final int[][] projectedPhysicalFields,
+ final List<String> metadataKeys) {
+ final Map<Integer, Integer> tableToDeserializedTopLevelPos =
+ tableToDeserializedTopLevelPos(physicalDataTypeProjection);
+
+ final List<int[]> deserializerProjectedFields = new ArrayList<>();
+ final Map<List<Integer>, Integer> deserializedToProducedPos = new
HashMap<>();
+ for (int producedPos = 0; producedPos <
projectedPhysicalFields.length; producedPos++) {
+ final int[] tablePos = projectedPhysicalFields[producedPos];
+ final int tableTopLevelPos = tablePos[0];
+
+ final Integer dataTypeTopLevelPos =
+ tableToDeserializedTopLevelPos.get(tableTopLevelPos);
+ if (dataTypeTopLevelPos != null) {
+ final int[] dataTypePos = copyArray(tablePos);
+ dataTypePos[0] = dataTypeTopLevelPos;
+
+ deserializerProjectedFields.add(dataTypePos);
+
+ final int deserializedPos = deserializerProjectedFields.size()
- 1;
+ deserializedToProducedPos.put(
+ Collections.singletonList(deserializedPos),
producedPos);
+ }
+ }
+
+ addMetadataProjections(
+ projectableDecodingFormat,
+ deserializerProjectedFields.size(),
+ projectedPhysicalFields.length,
+ metadataKeys,
+ deserializedToProducedPos);
+
+ return new Decoder(
+ projectableDecodingFormat.createRuntimeDecoder(
+ context,
+ toPhysicalDataType(
+ physicalTableDataType,
physicalDataTypeProjection, prefix),
+ deserializerProjectedFields.toArray(
+ new
int[deserializerProjectedFields.size()][])),
+ new ProjectorImpl(
+ deserializedToProducedPos,
+ deserializerProjectedFields.size(),
+ metadataKeys.size()));
+ }
+
+ /**
+ * This method generates a {@link Decoder} for a {@link
ProjectableDecodingFormat} that does not
+ * support <em>nested</em> projection. Only the required
<em>top-level</em> fields are pushed
+ * down into the format, and any nested sub-fields are extracted from the
deserialized row
+ * afterward by the {@link Projector}.
+ */
+ private static Decoder projectTopLevelInsideDeserializerThenNestedAfter(
+ final Context context,
+ final ProjectableDecodingFormat<DeserializationSchema<RowData>>
+ projectableDecodingFormat,
+ final DataType physicalTableDataType,
+ final int[] physicalDataTypeProjection,
+ final @Nullable String prefix,
+ final int[][] projectedPhysicalFields,
+ final List<String> metadataKeys) {
+ final Map<Integer, Integer> tableToDeserializedTopLevelPos =
+ tableToDeserializedTopLevelPos(physicalDataTypeProjection);
+
+ // Top-level fields (in data type space) to push into the format,
deduplicated so each
+ // required top-level field is only deserialized once.
+ final List<int[]> topLevelProjectedFields = new ArrayList<>();
+ final Map<Integer, Integer> dataTypeTopLevelToDeserializedPos = new
HashMap<>();
+ final Map<List<Integer>, Integer> deserializedToProducedPos = new
HashMap<>();
+ for (int producedPos = 0; producedPos <
projectedPhysicalFields.length; producedPos++) {
+ final int[] tablePos = projectedPhysicalFields[producedPos];
+ final int tableTopLevelPos = tablePos[0];
+
+ final Integer dataTypeTopLevelPos =
+ tableToDeserializedTopLevelPos.get(tableTopLevelPos);
+ if (dataTypeTopLevelPos != null) {
+ final int deserializedTopLevelPos =
+ dataTypeTopLevelToDeserializedPos.computeIfAbsent(
+ dataTypeTopLevelPos,
+ k -> {
+ topLevelProjectedFields.add(new int[] {k});
+ return topLevelProjectedFields.size() - 1;
+ });
+
+ // The remaining (nested) path is extracted from the
deserialized row afterward.
+ final int[] deserializedPos = copyArray(tablePos);
+ deserializedPos[0] = deserializedTopLevelPos;
+ deserializedToProducedPos.put(
+ Collections.unmodifiableList(
+ Arrays.stream(deserializedPos)
+ .boxed()
+ .collect(Collectors.toList())),
+ producedPos);
+ }
+ }
+
+ addMetadataProjections(
+ projectableDecodingFormat,
+ topLevelProjectedFields.size(),
+ projectedPhysicalFields.length,
+ metadataKeys,
+ deserializedToProducedPos);
+
+ return new Decoder(
+ projectableDecodingFormat.createRuntimeDecoder(
+ context,
+ toPhysicalDataType(
+ physicalTableDataType,
physicalDataTypeProjection, prefix),
+ topLevelProjectedFields.toArray(new
int[topLevelProjectedFields.size()][])),
+ new ProjectorImpl(
+ deserializedToProducedPos,
+ topLevelProjectedFields.size(),
+ metadataKeys.size()));
+ }
+
+ /**
+ * This method generates a {@link Decoder} which deserializes the data
fully using the {@link
+ * DecodingFormat} and then applies any projections afterward.
+ */
+ private static Decoder projectAfterDeserializing(
+ final Context context,
+ final DecodingFormat<DeserializationSchema<RowData>>
decodingFormat,
+ final DataType physicalTableDataType,
+ final int[] physicalDataTypeProjection,
+ final @Nullable String prefix,
+ final int[][] projectedPhysicalFields,
+ final List<String> metadataKeys) {
+ final DataType physicalDataType =
+ toPhysicalDataType(physicalTableDataType,
physicalDataTypeProjection, prefix);
+ final Map<Integer, Integer> tableToDeserializedTopLevelPos =
+ tableToDeserializedTopLevelPos(physicalDataTypeProjection);
+
+ final Map<List<Integer>, Integer> deserializedToProducedPos = new
HashMap<>();
+ for (int producedPos = 0; producedPos <
projectedPhysicalFields.length; producedPos++) {
+ final int[] tablePos = projectedPhysicalFields[producedPos];
+ int tableTopLevelPos = tablePos[0];
+
+ final Integer deserializedTopLevelPos =
+ tableToDeserializedTopLevelPos.get(tableTopLevelPos);
+ if (deserializedTopLevelPos != null) {
+ final int[] deserializedPos = copyArray(tablePos);
+ deserializedPos[0] = deserializedTopLevelPos;
+
+ deserializedToProducedPos.put(
+ Collections.unmodifiableList(
+ Arrays.stream(deserializedPos)
+ .boxed()
+ .collect(Collectors.toList())),
+ producedPos);
+ }
+ }
+
+ addMetadataProjections(
+ decodingFormat,
+ physicalDataTypeProjection.length,
+ projectedPhysicalFields.length,
+ metadataKeys,
+ deserializedToProducedPos);
+
+ return new Decoder(
+ decodingFormat.createRuntimeDecoder(context, physicalDataType),
+ new ProjectorImpl(
+ deserializedToProducedPos,
+ physicalDataTypeProjection.length,
+ metadataKeys.size()));
+ }
+
+ /** Projects fields from the deserialized row to their positions in the
final produced row. */
+ @Internal
+ public interface Projector extends Serializable {
+ /** Returns true if {@link #project} will not project any fields. */
+ boolean isEmptyProjection();
+
+ /**
+ * Returns true if projection is needed i.e. if the produced record is
different from the
+ * deserialized record
+ */
+ boolean isProjectionNeeded();
+
+ /** Copies fields from the deserialized row to their final positions
in the produced row. */
+ void project(final RowData deserialized, final GenericRowData
producedRow);
+ }
+
+ private static class ProjectorImpl implements Projector {
+
+ private final Map<List<Integer>, Integer> deserializedToProducedPos;
+ private final boolean isProjectionNeeded;
+
+ ProjectorImpl(
+ final Map<List<Integer>, Integer> deserializedToProducedPos,
+ final int numDeserializedPhysicalFields,
+ final int numMetadataFields) {
+ this.deserializedToProducedPos = deserializedToProducedPos;
+ this.isProjectionNeeded =
+ !(deserializedToProducedPos.size()
+ == (numDeserializedPhysicalFields +
numMetadataFields)
+ && samePositions(deserializedToProducedPos));
+ }
+
+ private static boolean samePositions(
+ Map<List<Integer>, Integer> deserializedToProducedPos) {
+ return deserializedToProducedPos.entrySet().stream()
+ .allMatch(
+ entry -> {
+ final List<Integer> deserializedPos =
entry.getKey();
+ final List<Integer> producedPos =
+
Collections.singletonList(entry.getValue());
+ return Objects.equals(producedPos,
deserializedPos);
+ });
+ }
+
+ @Override
+ public boolean isEmptyProjection() {
+ return deserializedToProducedPos.isEmpty();
+ }
+
+ @Override
+ public boolean isProjectionNeeded() {
+ return isProjectionNeeded;
+ }
+
+ @Override
+ public void project(final RowData deserialized, final GenericRowData
producedRow) {
Review Comment:
This runs per record for every table with a key or metadata, and by default
for any narrowing query. The map walk allocates 24 to 160 bytes a record where
the `int[]` loop it replaces allocated none, a flat `int[]` path for the top
level entries avoids that.
--
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]
