[mysql] Support metadata columns for mysql-cdc connector (#496)
Jark Wu
3 years ago
committed by
GitHub
parent
64fc3e34b5
commit
12f27acda2
@ -0,0 +1,60 @@
|
||||
/*
|
||||
* 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 com.ververica.cdc.debezium.table;
|
||||
|
||||
import org.apache.flink.annotation.Internal;
|
||||
import org.apache.flink.table.data.GenericRowData;
|
||||
import org.apache.flink.table.data.RowData;
|
||||
import org.apache.flink.table.data.utils.JoinedRowData;
|
||||
import org.apache.flink.util.Collector;
|
||||
|
||||
import org.apache.kafka.connect.source.SourceRecord;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/** Emits a row with physical fields and metadata fields. */
|
||||
@Internal
|
||||
public final class AppendMetadataCollector implements Collector<RowData>, Serializable {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private final MetadataConverter[] metadataConverters;
|
||||
|
||||
public transient SourceRecord inputRecord;
|
||||
public transient Collector<RowData> outputCollector;
|
||||
|
||||
public AppendMetadataCollector(MetadataConverter[] metadataConverters) {
|
||||
this.metadataConverters = metadataConverters;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void collect(RowData physicalRow) {
|
||||
GenericRowData metaRow = new GenericRowData(metadataConverters.length);
|
||||
for (int i = 0; i < metadataConverters.length; i++) {
|
||||
Object meta = metadataConverters[i].read(inputRecord);
|
||||
metaRow.setField(i, meta);
|
||||
}
|
||||
RowData outRow = new JoinedRowData(physicalRow.getRowKind(), physicalRow, metaRow);
|
||||
outputCollector.collect(outRow);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
// nothing to do
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,32 @@
|
||||
/*
|
||||
* 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 com.ververica.cdc.debezium.table;
|
||||
|
||||
import org.apache.flink.annotation.Internal;
|
||||
|
||||
import org.apache.kafka.connect.source.SourceRecord;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/** A converter converts {@link SourceRecord} metadata into Flink internal data structures. */
|
||||
@FunctionalInterface
|
||||
@Internal
|
||||
public interface MetadataConverter extends Serializable {
|
||||
Object read(SourceRecord record);
|
||||
}
|
||||
@ -0,0 +1,108 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.ververica.cdc.connectors.mysql.table;
|
||||
|
||||
import org.apache.flink.table.api.DataTypes;
|
||||
import org.apache.flink.table.data.StringData;
|
||||
import org.apache.flink.table.data.TimestampData;
|
||||
import org.apache.flink.table.types.DataType;
|
||||
|
||||
import com.ververica.cdc.debezium.table.MetadataConverter;
|
||||
import io.debezium.connector.AbstractSourceInfo;
|
||||
import io.debezium.data.Envelope;
|
||||
import org.apache.kafka.connect.data.Struct;
|
||||
import org.apache.kafka.connect.source.SourceRecord;
|
||||
|
||||
/** Defines the supported metadata columns for {@link MySqlTableSource}. */
|
||||
public enum MySqlReadableMetadata {
|
||||
/** Name of the table that contain the row. . */
|
||||
TABLE_NAME(
|
||||
"table_name",
|
||||
DataTypes.STRING().notNull(),
|
||||
new MetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(SourceRecord record) {
|
||||
Struct messageStruct = (Struct) record.value();
|
||||
Struct sourceStruct = messageStruct.getStruct(Envelope.FieldName.SOURCE);
|
||||
return StringData.fromString(
|
||||
sourceStruct.getString(AbstractSourceInfo.TABLE_NAME_KEY));
|
||||
}
|
||||
}),
|
||||
|
||||
/** Name of the database that contain the row. */
|
||||
DATABASE_NAME(
|
||||
"database_name",
|
||||
DataTypes.STRING().notNull(),
|
||||
new MetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(SourceRecord record) {
|
||||
Struct messageStruct = (Struct) record.value();
|
||||
Struct sourceStruct = messageStruct.getStruct(Envelope.FieldName.SOURCE);
|
||||
return StringData.fromString(
|
||||
sourceStruct.getString(AbstractSourceInfo.DATABASE_NAME_KEY));
|
||||
}
|
||||
}),
|
||||
|
||||
/**
|
||||
* It indicates the time that the change was made in the database. If the record is read from
|
||||
* snapshot of the table instead of the binlog, the value is always 0.
|
||||
*/
|
||||
OP_TS(
|
||||
"op_ts",
|
||||
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3).notNull(),
|
||||
new MetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(SourceRecord record) {
|
||||
Struct messageStruct = (Struct) record.value();
|
||||
Struct sourceStruct = messageStruct.getStruct(Envelope.FieldName.SOURCE);
|
||||
return TimestampData.fromEpochMillis(
|
||||
(Long) sourceStruct.get(AbstractSourceInfo.TIMESTAMP_KEY));
|
||||
}
|
||||
});
|
||||
|
||||
private final String key;
|
||||
|
||||
private final DataType dataType;
|
||||
|
||||
private final MetadataConverter converter;
|
||||
|
||||
MySqlReadableMetadata(String key, DataType dataType, MetadataConverter converter) {
|
||||
this.key = key;
|
||||
this.dataType = dataType;
|
||||
this.converter = converter;
|
||||
}
|
||||
|
||||
public String getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
public DataType getDataType() {
|
||||
return dataType;
|
||||
}
|
||||
|
||||
public MetadataConverter getConverter() {
|
||||
return converter;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue