leonardBang commented on code in PR #4494: URL: https://github.com/apache/flink-cdc/pull/4494#discussion_r3820591573
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/deserializer/FlussRecordDeserializer.java: ########## @@ -0,0 +1,504 @@ +/* + * 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.cdc.connectors.fluss.source.deserializer; + +import org.apache.flink.cdc.common.data.DecimalData; +import org.apache.flink.cdc.common.data.GenericArrayData; +import org.apache.flink.cdc.common.data.GenericMapData; +import org.apache.flink.cdc.common.data.GenericRecordData; +import org.apache.flink.cdc.common.data.LocalZonedTimestampData; +import org.apache.flink.cdc.common.data.RecordData; +import org.apache.flink.cdc.common.data.TimestampData; +import org.apache.flink.cdc.common.data.binary.BinaryStringData; +import org.apache.flink.cdc.common.event.AddColumnEvent; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.schema.Column; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.connectors.fluss.source.reader.FlussSourceRecord; +import org.apache.flink.cdc.connectors.fluss.utils.FlussConversions; +import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator; + +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.Decimal; +import org.apache.fluss.row.InternalArray; +import org.apache.fluss.row.InternalMap; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.ProjectedRow; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.types.ArrayType; +import org.apache.fluss.types.DataField; +import org.apache.fluss.types.MapType; +import org.apache.fluss.types.RowType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * A CDC-specific implementation of {@link FlussDeserializer} that converts Fluss {@link + * ScanRecord}s into Flink CDC {@link Event}s (DataChangeEvents). + * + * <p>This class maps Fluss ChangeType to the appropriate CDC operation type (INSERT, UPDATE, + * DELETE). + */ +public class FlussRecordDeserializer implements FlussDeserializer<Event> { + private static final Logger LOG = LoggerFactory.getLogger(FlussRecordDeserializer.class); + + private static final long serialVersionUID = 1L; + + /** Cache of the last-seen schemaId per table (log records only). */ + private transient Map<TablePath, Integer> latestSchemaIdCache; + + /** Cache of the last-seen RowType per table, used to detect schema changes. */ + private transient Map<TablePath, RowType> latestRowTypeCache; + + /** Cache of row data generators per table. */ + private transient Map<TablePath, BinaryRecordDataGenerator> latestRecordDataGeneratorCache; + + /** Cache of field converters per table, used to avoid rebuilding nested type converters. */ + private transient Map<TablePath, FlussDeserializationConverter[]> latestFieldConverterCache; + + /** Tables restored from split state whose CreateTableEvent needs fresh table key metadata. */ + private transient Map<TablePath, RowType> restoredCreateTableRowTypeCache; + + @Override + public List<Event> deserialize(FlussSourceRecord record, TablePath tablePath) { + List<Event> events = new ArrayList<>(); + TableId tableId = TableId.tableId(tablePath.getDatabaseName(), tablePath.getTableName()); + RowType rowType = record.getRowType(); + + boolean isSchemaChangeEvent = inferSchemaChangeEvent(events, record, tablePath, tableId); + InternalRow row = record.getScanRecord().getRow(); + ChangeType changeType = record.getScanRecord().getChangeType(); + + switch (changeType) { + case APPEND_ONLY: + case INSERT: + { + RecordData after = + convertFlussRowToCdcRecord( + tablePath, row, rowType, isSchemaChangeEvent); + events.add(DataChangeEvent.insertEvent(tableId, after)); + break; + } + case UPDATE_BEFORE: + // UPDATE_BEFORE is typically followed by UPDATE_AFTER. + // We skip it here and handle the full update via UPDATE_AFTER. + break; + case UPDATE_AFTER: + { + RecordData after = + convertFlussRowToCdcRecord( + tablePath, row, rowType, isSchemaChangeEvent); + events.add(DataChangeEvent.replaceEvent(tableId, after)); + break; + } + case DELETE: + { + RecordData before = + convertFlussRowToCdcRecord( + tablePath, row, rowType, isSchemaChangeEvent); + events.add(DataChangeEvent.deleteEvent(tableId, before)); + break; + } + default: + throw new IllegalArgumentException("Unsupported change type: " + changeType); + } + return events; + } + + private boolean inferSchemaChangeEvent( + List<Event> events, FlussSourceRecord record, TablePath tablePath, TableId tableId) { + // Detect schema changes for log records (schemaId >= 0). + // Snapshot records have schemaId = -1 and are skipped. + boolean inferSchemaChangeEvent = false; + int schemaId = record.getScanRecord().getSchemaId(); + RowType rowType = record.getRowType(); + org.apache.flink.cdc.common.types.RowType cdcRowType = + (org.apache.flink.cdc.common.types.RowType) FlussConversions.toCdcType(rowType); + if (schemaId >= 0) { + ensureCacheInitialized(); + RowType restoredRowType = restoredCreateTableRowTypeCache.remove(tablePath); + if (restoredRowType != null) { + events.add( + new CreateTableEvent( + tableId, + buildCdcSchema( + restoredRowType, + record.getPrimaryKeyNames(), + record.getPartitionKeyNames()))); + } + + Integer cachedSchemaId = latestSchemaIdCache.get(tablePath); Review Comment: Could we restore the schema with the highest schema ID here instead of keeping whichever split is processed first? Different splits of the same table may checkpoint different schema IDs, and the order in which restored splits are applied is not guaranteed to be newest-first. Keeping the first schema can therefore initialize the table cache with an older schema and may emit an outdated CreateTableEvent. It may be helpful to add restore tests for both split orders {schema 1, schema 2} and {schema 2, schema 1}. -- 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]
