wuchong commented on a change in pull request #13862:
URL: https://github.com/apache/flink/pull/13862#discussion_r515829181
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##########
@@ -102,6 +108,36 @@
.withDescription("Defines the format identifier for encoding
data. " +
"The identifier is used to discover a suitable format
factory.");
+ public static final ConfigOption<List<String>> KEY_FIELDS =
ConfigOptions
Review comment:
Could these options belong to `KafkaOptions`?
IMO, they are very Kafka specific features and can't reused by most other
connectors (e.g. Kinesis).
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java
##########
@@ -131,18 +156,22 @@ public KafkaDynamicSource(
@Override
public ChangelogMode getChangelogMode() {
- return this.decodingFormat.getChangelogMode();
+ return valueDecodingFormat.getChangelogMode();
}
@Override
- public ScanRuntimeProvider getScanRuntimeProvider(ScanContext
runtimeProviderContext) {
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
+ final DeserializationSchema<RowData> keyDeserialization =
+ createDeserialization(context,
keyDecodingFormat, keyProjection, keyPrefix);
+
final DeserializationSchema<RowData> valueDeserialization =
-
decodingFormat.createRuntimeDecoder(runtimeProviderContext, physicalDataType);
+ createDeserialization(context,
valueDecodingFormat, valueProjection, keyPrefix);
Review comment:
ditto.
##########
File path:
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -383,6 +383,147 @@ public void testKafkaSourceSinkWithMetadata() throws
Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithKeyAndPartialValue() throws
Exception {
+ if (isLegacyConnector) {
+ return;
+ }
+ // we always use a different topic name for each parameterized
topic,
+ // in order to make sure the topic can be created.
+ final String topic = "key_partial_value_topic_" + format;
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka
-------------------
+ String groupId = standardProps.getProperty("group.id");
+ String bootstraps =
standardProps.getProperty("bootstrap.servers");
+
+ // k_user_id and user_id have different data types to verify
the correct mapping,
+ // fields are reordered on purpose
+ final String createTable = String.format(
+ "CREATE TABLE kafka (\n"
+ + " `k_user_id` BIGINT,\n"
+ + " `name` STRING,\n"
+ + " `timestamp` TIMESTAMP(3)
METADATA,\n"
+ + " `k_event_id` BIGINT,\n"
+ + " `user_id` INT,\n"
+ + " `payload` 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"
+ + " 'key.format' = '%s',\n"
+ + " 'key.fields' =
'k_event_id; k_user_id',\n"
+ + " 'key.fields-prefix' =
'k_',\n"
+ + " 'value.format' = '%s',\n"
+ + " 'value.fields-include' =
'EXCEPT_KEY'\n"
+ + ")",
+ topic,
+ bootstraps,
+ groupId,
+ format,
+ format);
+
+ tEnv.executeSql(createTable);
+
+ String initialValues = "INSERT INTO kafka\n"
+ + "VALUES\n"
+ + " (1, 'name 1', TIMESTAMP '2020-03-08
13:12:11.123', 100, 1, 'payload 1'),\n"
+ + " (2, 'name 2', TIMESTAMP '2020-03-09
13:12:11.123', 101, 2, 'payload 2'),\n"
+ + " (3, 'name 3', TIMESTAMP '2020-03-10
13:12:11.123', 102, 3, 'payload 3')";
Review comment:
minor: how about having differnet values for `user_id`? This can help
to verify we can extract correct values when name clashes.
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
##########
@@ -129,78 +142,183 @@ public DynamicTableSource
createDynamicTableSource(Context context) {
final DataType physicalDataType =
context.getCatalogTable().getSchema().toPhysicalRowDataType();
+ final int[] keyProjection =
FactoryUtil.createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final int[] valueProjection =
FactoryUtil.createValueFormatProjection(tableOptions, physicalDataType);
+
+ final String keyPrefix =
tableOptions.getOptional(FactoryUtil.KEY_FIELDS_PREFIX).orElse(null);
+
return createKafkaTableSource(
Review comment:
We don't need to support different Kafka versions now. Therefore, I
think we can directly construct the `KafkaDynamicSource` and `KafkaDynamicSink`
here without the abstracted `createKafkaTableSource` method.
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
##########
@@ -83,38 +97,53 @@
protected final Properties properties;
/** Partitioner to select Kafka partition for each item. */
- protected final Optional<FlinkKafkaPartitioner<RowData>> partitioner;
+ protected final @Nullable FlinkKafkaPartitioner<RowData> partitioner;
/** Sink commit semantic.*/
protected final KafkaSinkSemantic semantic;
public KafkaDynamicSink(
DataType physicalDataType,
+ @Nullable EncodingFormat<SerializationSchema<RowData>>
keyEncodingFormat,
+ EncodingFormat<SerializationSchema<RowData>>
valueEncodingFormat,
+ int[] keyProjection,
+ int[] valueProjection,
+ @Nullable String keyPrefix,
String topic,
Properties properties,
- Optional<FlinkKafkaPartitioner<RowData>> partitioner,
- EncodingFormat<SerializationSchema<RowData>>
encodingFormat,
+ @Nullable FlinkKafkaPartitioner<RowData> partitioner,
KafkaSinkSemantic semantic) {
+ // Format attributes
this.physicalDataType =
Preconditions.checkNotNull(physicalDataType, "Physical data type must not be
null.");
+ this.keyEncodingFormat = keyEncodingFormat;
+ this.valueEncodingFormat =
Preconditions.checkNotNull(valueEncodingFormat, "Value encoding format must not
be null.");
+ this.keyProjection = Preconditions.checkNotNull(keyProjection,
"Key projection must not be null.");
+ this.valueProjection =
Preconditions.checkNotNull(valueProjection, "Value projection must not be
null.");
+ this.keyPrefix = keyPrefix;
+ // Mutable attributes
this.metadataKeys = Collections.emptyList();
+ // Kafka-specific attributes
this.topic = Preconditions.checkNotNull(topic, "Topic must not
be null.");
this.properties = Preconditions.checkNotNull(properties,
"Properties must not be null.");
- this.partitioner = Preconditions.checkNotNull(partitioner,
"Partitioner must not be null.");
- this.encodingFormat =
Preconditions.checkNotNull(encodingFormat, "Encoding format must not be null.");
+ this.partitioner = partitioner;
this.semantic = Preconditions.checkNotNull(semantic, "Semantic
must not be null.");
}
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
- return this.encodingFormat.getChangelogMode();
+ return valueEncodingFormat.getChangelogMode();
}
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ final SerializationSchema<RowData> keySerialization =
+ createSerialization(context, keyEncodingFormat,
keyProjection, keyPrefix);
+
final SerializationSchema<RowData> valueSerialization =
-
this.encodingFormat.createRuntimeEncoder(context, this.physicalDataType);
+ createSerialization(context,
valueEncodingFormat, valueProjection, keyPrefix);
Review comment:
Should use `null` for `prefix` parameter? IIUC, the `keyPrefix` should
only affects key de/serialization. No matter the key fields are included or
excluded in the value fields, they shoudn't be stripped by prefix when building
value de/serialization. For example:
```sql
CREATE TABLE kafka (
key_id BIGINT,
user_id BIGINT,
key_name STRING,
payload STRING
) WITH (
'connector' = 'kafka',
'key.fields' = 'key_id',
'key.fields-prefix' = 'key',
'value.fields-include' = 'EXCEPT_KEY'
)
```
The RowType of value format should be `user_id, key_name, payload` instead
of `user_id, name, payload`, right?
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
##########
@@ -129,78 +142,183 @@ public DynamicTableSource
createDynamicTableSource(Context context) {
final DataType physicalDataType =
context.getCatalogTable().getSchema().toPhysicalRowDataType();
+ final int[] keyProjection =
FactoryUtil.createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final int[] valueProjection =
FactoryUtil.createValueFormatProjection(tableOptions, physicalDataType);
+
+ final String keyPrefix =
tableOptions.getOptional(FactoryUtil.KEY_FIELDS_PREFIX).orElse(null);
+
return createKafkaTableSource(
physicalDataType,
+ keyDecodingFormat.orElse(null),
+ valueDecodingFormat,
+ keyProjection,
+ valueProjection,
+ keyPrefix,
KafkaOptions.getSourceTopics(tableOptions),
KafkaOptions.getSourceTopicPattern(tableOptions),
properties,
- decodingFormat,
startupOptions.startupMode,
startupOptions.specificOffsets,
startupOptions.startupTimestampMillis);
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
- FactoryUtil.TableFactoryHelper helper =
FactoryUtil.createTableFactoryHelper(this, context);
+ final TableFactoryHelper helper =
FactoryUtil.createTableFactoryHelper(this, context);
- ReadableConfig tableOptions = helper.getOptions();
+ final ReadableConfig tableOptions = helper.getOptions();
- EncodingFormat<SerializationSchema<RowData>> encodingFormat =
helper.discoverEncodingFormat(
- SerializationFormatFactory.class,
- FactoryUtil.FORMAT);
+ final Optional<EncodingFormat<SerializationSchema<RowData>>>
keyEncodingFormat =
+ getKeyEncodingFormat(helper);
+
+ final EncodingFormat<SerializationSchema<RowData>>
valueEncodingFormat =
+ getValueEncodingFormat(helper,
keyEncodingFormat.isPresent());
- // Validate the option values.
- validateTableSinkOptions(tableOptions);
- // Validate the option data type.
helper.validateExcept(PROPERTIES_PREFIX);
+ validateTableSinkOptions(tableOptions);
+
final DataType physicalDataType =
context.getCatalogTable().getSchema().toPhysicalRowDataType();
+ final int[] keyProjection =
FactoryUtil.createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final int[] valueProjection =
FactoryUtil.createValueFormatProjection(tableOptions, physicalDataType);
+
+ final String keyPrefix =
tableOptions.getOptional(FactoryUtil.KEY_FIELDS_PREFIX).orElse(null);
+
return createKafkaTableSink(
physicalDataType,
+ keyEncodingFormat.orElse(null),
+ valueEncodingFormat,
+ keyProjection,
+ valueProjection,
+ keyPrefix,
tableOptions.get(TOPIC).get(0),
getKafkaProperties(context.getCatalogTable().getOptions()),
- getFlinkKafkaPartitioner(tableOptions,
context.getClassLoader()),
- encodingFormat,
+ getFlinkKafkaPartitioner(tableOptions,
context.getClassLoader()).orElse(null),
getSinkSemantic(tableOptions));
}
//
--------------------------------------------------------------------------------------------
+ private static Optional<DecodingFormat<DeserializationSchema<RowData>>>
getKeyDecodingFormat(
+ TableFactoryHelper helper) {
+ final Optional<DecodingFormat<DeserializationSchema<RowData>>>
keyDecodingFormat =
+ helper.discoverOptionalDecodingFormat(
+
DeserializationFormatFactory.class,
+ FactoryUtil.KEY_FORMAT);
+ keyDecodingFormat.ifPresent(format -> {
+ if
(!format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "A key format
should only deal with INSERT-only records. "
+
+ "But %s has a changelog mode of %s.",
+
helper.getOptions().get(FactoryUtil.KEY_FORMAT),
+
format.getChangelogMode()));
+ }
+ });
+ return keyDecodingFormat;
+ }
+
+ private static Optional<EncodingFormat<SerializationSchema<RowData>>>
getKeyEncodingFormat(
+ TableFactoryHelper helper) {
+ final Optional<EncodingFormat<SerializationSchema<RowData>>>
keyEncodingFormat =
+ helper.discoverOptionalEncodingFormat(
+
SerializationFormatFactory.class,
+ FactoryUtil.KEY_FORMAT);
+ keyEncodingFormat.ifPresent(format -> {
+ if
(!format.getChangelogMode().containsOnly(RowKind.INSERT)) {
+ throw new ValidationException(
+ String.format(
+ "A key format
should only deal with INSERT-only records. "
+
+ "But %s has a changelog mode of %s.",
+
helper.getOptions().get(FactoryUtil.KEY_FORMAT),
+
format.getChangelogMode()));
+ }
+ });
+ return keyEncodingFormat;
+ }
+
+ private static DecodingFormat<DeserializationSchema<RowData>>
getValueDecodingFormat(
+ TableFactoryHelper helper,
+ boolean hasKeyFormat) {
+ final ConfigOption<String> valueFormatOption;
+ if (hasKeyFormat) {
+ valueFormatOption = FactoryUtil.VALUE_FORMAT;
+ } else {
+ valueFormatOption = FactoryUtil.FORMAT;
+ }
Review comment:
Is this as expected? Could users use `value.format` without
`key.format`? I thought `value.format` is a repalcement of `format`.
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
##########
@@ -83,38 +97,53 @@
protected final Properties properties;
/** Partitioner to select Kafka partition for each item. */
- protected final Optional<FlinkKafkaPartitioner<RowData>> partitioner;
+ protected final @Nullable FlinkKafkaPartitioner<RowData> partitioner;
/** Sink commit semantic.*/
protected final KafkaSinkSemantic semantic;
public KafkaDynamicSink(
DataType physicalDataType,
+ @Nullable EncodingFormat<SerializationSchema<RowData>>
keyEncodingFormat,
+ EncodingFormat<SerializationSchema<RowData>>
valueEncodingFormat,
+ int[] keyProjection,
+ int[] valueProjection,
+ @Nullable String keyPrefix,
String topic,
Properties properties,
- Optional<FlinkKafkaPartitioner<RowData>> partitioner,
- EncodingFormat<SerializationSchema<RowData>>
encodingFormat,
+ @Nullable FlinkKafkaPartitioner<RowData> partitioner,
KafkaSinkSemantic semantic) {
+ // Format attributes
this.physicalDataType =
Preconditions.checkNotNull(physicalDataType, "Physical data type must not be
null.");
+ this.keyEncodingFormat = keyEncodingFormat;
+ this.valueEncodingFormat =
Preconditions.checkNotNull(valueEncodingFormat, "Value encoding format must not
be null.");
+ this.keyProjection = Preconditions.checkNotNull(keyProjection,
"Key projection must not be null.");
+ this.valueProjection =
Preconditions.checkNotNull(valueProjection, "Value projection must not be
null.");
+ this.keyPrefix = keyPrefix;
+ // Mutable attributes
this.metadataKeys = Collections.emptyList();
+ // Kafka-specific attributes
this.topic = Preconditions.checkNotNull(topic, "Topic must not
be null.");
this.properties = Preconditions.checkNotNull(properties,
"Properties must not be null.");
- this.partitioner = Preconditions.checkNotNull(partitioner,
"Partitioner must not be null.");
- this.encodingFormat =
Preconditions.checkNotNull(encodingFormat, "Encoding format must not be null.");
+ this.partitioner = partitioner;
this.semantic = Preconditions.checkNotNull(semantic, "Semantic
must not be null.");
}
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
- return this.encodingFormat.getChangelogMode();
+ return valueEncodingFormat.getChangelogMode();
}
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ final SerializationSchema<RowData> keySerialization =
+ createSerialization(context, keyEncodingFormat,
keyProjection, keyPrefix);
+
final SerializationSchema<RowData> valueSerialization =
-
this.encodingFormat.createRuntimeEncoder(context, this.physicalDataType);
+ createSerialization(context,
valueEncodingFormat, valueProjection, keyPrefix);
Review comment:
What do you think about:
when `'key.fields-prefix'` is defined, `value.fields-include` shoudn't be
defined to `ALL`, and all value fields shoudn't be prefixed with `keyPrefix`.
I think `'key.fields-prefix'` is only used when users want read out key and
value parts separately and avoid name clashes.
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
##########
@@ -129,78 +142,183 @@ public DynamicTableSource
createDynamicTableSource(Context context) {
final DataType physicalDataType =
context.getCatalogTable().getSchema().toPhysicalRowDataType();
+ final int[] keyProjection =
FactoryUtil.createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final int[] valueProjection =
FactoryUtil.createValueFormatProjection(tableOptions, physicalDataType);
+
+ final String keyPrefix =
tableOptions.getOptional(FactoryUtil.KEY_FIELDS_PREFIX).orElse(null);
Review comment:
Could we add a validation that the declared key fields should be all
prefixed with the keyPrefix?
##########
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##########
@@ -102,6 +108,36 @@
.withDescription("Defines the format identifier for encoding
data. " +
"The identifier is used to discover a suitable format
factory.");
+ public static final ConfigOption<List<String>> KEY_FIELDS =
ConfigOptions
Review comment:
Yes, personally, I think they should be moved to Kafka and TestUtils
too.
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java
##########
@@ -129,78 +142,183 @@ public DynamicTableSource
createDynamicTableSource(Context context) {
final DataType physicalDataType =
context.getCatalogTable().getSchema().toPhysicalRowDataType();
+ final int[] keyProjection =
FactoryUtil.createKeyFormatProjection(tableOptions, physicalDataType);
+
+ final int[] valueProjection =
FactoryUtil.createValueFormatProjection(tableOptions, physicalDataType);
+
+ final String keyPrefix =
tableOptions.getOptional(FactoryUtil.KEY_FIELDS_PREFIX).orElse(null);
+
return createKafkaTableSource(
Review comment:
I see.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
