[tidb] Support metadata column
eastfisher
3 years ago
committed by
Leonard Xu
parent
570cba1159
commit
a728c89937
@ -0,0 +1,61 @@
|
||||
/*
|
||||
* 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.tidb.table;
|
||||
|
||||
import org.apache.flink.table.data.RowData;
|
||||
import org.apache.flink.util.Collector;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
import static org.apache.flink.util.Preconditions.checkNotNull;
|
||||
|
||||
/**
|
||||
* Base class of deserialization schema from TiKV RowValue (Snapshot or Change Event) to Flink
|
||||
* Table/SQL internal data structure {@link RowData}.
|
||||
*/
|
||||
public class RowDataTiKVEventDeserializationSchemaBase implements Serializable {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
/** Whether the deserializer needs to handle metadata columns. */
|
||||
private final boolean hasMetadata;
|
||||
|
||||
/**
|
||||
* A wrapped output collector which is used to append metadata columns after physical columns.
|
||||
*/
|
||||
private final TiKVAppendMetadataCollector appendMetadataCollector;
|
||||
|
||||
public RowDataTiKVEventDeserializationSchemaBase(TiKVMetadataConverter[] metadataConverters) {
|
||||
this.hasMetadata = checkNotNull(metadataConverters).length > 0;
|
||||
this.appendMetadataCollector = new TiKVAppendMetadataCollector(metadataConverters);
|
||||
}
|
||||
|
||||
public void emit(
|
||||
TiKVMetadataConverter.TiKVRowValue inRecord,
|
||||
RowData physicalRow,
|
||||
Collector<RowData> collector) {
|
||||
if (!hasMetadata) {
|
||||
collector.collect(physicalRow);
|
||||
return;
|
||||
}
|
||||
|
||||
appendMetadataCollector.row = inRecord;
|
||||
appendMetadataCollector.outputCollector = collector;
|
||||
appendMetadataCollector.collect(physicalRow);
|
||||
}
|
||||
}
|
||||
@ -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 com.ververica.cdc.connectors.tidb.table;
|
||||
|
||||
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 java.io.Serializable;
|
||||
|
||||
/** Emits a row with physical fields and metadata fields. */
|
||||
public class TiKVAppendMetadataCollector implements Collector<RowData>, Serializable {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
private final TiKVMetadataConverter[] metadataConverters;
|
||||
|
||||
public transient TiKVMetadataConverter.TiKVRowValue row;
|
||||
public transient Collector<RowData> outputCollector;
|
||||
|
||||
public TiKVAppendMetadataCollector(TiKVMetadataConverter[] 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(row);
|
||||
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,50 @@
|
||||
/*
|
||||
* 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.tidb.table;
|
||||
|
||||
import org.apache.flink.annotation.Internal;
|
||||
|
||||
import org.tikv.kvproto.Cdcpb;
|
||||
import org.tikv.kvproto.Kvrpcpb;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/** A converter converts TiKV Row metadata into Flink internal data structures. */
|
||||
@FunctionalInterface
|
||||
@Internal
|
||||
public interface TiKVMetadataConverter extends Serializable {
|
||||
Object read(TiKVRowValue row);
|
||||
|
||||
/** TiKV Row Value. */
|
||||
class TiKVRowValue {
|
||||
public boolean isKv;
|
||||
public Kvrpcpb.KvPair kvPair;
|
||||
public Cdcpb.Event.Row row;
|
||||
|
||||
public TiKVRowValue(Kvrpcpb.KvPair kvPair) {
|
||||
this.isKv = true;
|
||||
this.kvPair = kvPair;
|
||||
}
|
||||
|
||||
public TiKVRowValue(Cdcpb.Event.Row row) {
|
||||
this.isKv = false;
|
||||
this.row = row;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,113 @@
|
||||
/*
|
||||
* 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.tidb.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 java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/** Defines the supported metadata columns for {@link TiDBTableSource}. */
|
||||
public class TiKVReadableMetadata {
|
||||
|
||||
private final String key;
|
||||
|
||||
private final DataType dataType;
|
||||
|
||||
private final TiKVMetadataConverter converter;
|
||||
|
||||
TiKVReadableMetadata(String key, DataType dataType, TiKVMetadataConverter converter) {
|
||||
this.key = key;
|
||||
this.dataType = dataType;
|
||||
this.converter = converter;
|
||||
}
|
||||
|
||||
public String getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
public DataType getDataType() {
|
||||
return dataType;
|
||||
}
|
||||
|
||||
public TiKVMetadataConverter getConverter() {
|
||||
return converter;
|
||||
}
|
||||
|
||||
/** Name of the table that contain the row. */
|
||||
public static TiKVReadableMetadata createTableNameMetadata(String tableName) {
|
||||
return new TiKVReadableMetadata(
|
||||
"table_name",
|
||||
DataTypes.STRING().notNull(),
|
||||
new TiKVMetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(TiKVRowValue row) {
|
||||
return StringData.fromString(tableName);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Name of the database that contain the row. */
|
||||
public static TiKVReadableMetadata createDatabaseNameMetadata(String database) {
|
||||
return new TiKVReadableMetadata(
|
||||
"database_name",
|
||||
DataTypes.STRING().notNull(),
|
||||
new TiKVMetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(TiKVRowValue row) {
|
||||
return StringData.fromString(database);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public static TiKVReadableMetadata createOpTsMetadata() {
|
||||
return new TiKVReadableMetadata(
|
||||
"op_ts",
|
||||
DataTypes.TIMESTAMP_LTZ(3).notNull(),
|
||||
new TiKVMetadataConverter() {
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public Object read(TiKVRowValue row) {
|
||||
if (row.isKv) {
|
||||
// We cannot get ts from KvPair, use default value.
|
||||
return TimestampData.fromEpochMillis(0);
|
||||
} else {
|
||||
return TimestampData.fromEpochMillis(row.row.getStartTs());
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public static TiKVReadableMetadata[] createTiKVReadableMetadata(
|
||||
String database, String tableName) {
|
||||
List<TiKVReadableMetadata> list = new ArrayList<>();
|
||||
list.add(createDatabaseNameMetadata(database));
|
||||
list.add(createTableNameMetadata(tableName));
|
||||
list.add(createOpTsMetadata());
|
||||
return list.toArray(new TiKVReadableMetadata[0]);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue