leonardBang commented on code in PR #4494: URL: https://github.com/apache/flink-cdc/pull/4494#discussion_r3821025836
########## 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); + if (cachedSchemaId == null || schemaId > cachedSchemaId) { + if (cachedSchemaId == null) { + // First record for this table — emit CreateTableEvent with table keys. + events.add( + new CreateTableEvent( + tableId, + buildCdcSchema( + rowType, + record.getPrimaryKeyNames(), + record.getPartitionKeyNames()))); + } else { + // SchemaId changed — infer and emit schema change events + inferSchemaChangeEvent = true; + RowType oldRowType = latestRowTypeCache.get(tablePath); + events.addAll(inferSchemaChanges(tableId, tablePath, oldRowType, rowType)); + } + latestSchemaIdCache.put(tablePath, schemaId); + latestRowTypeCache.put(tablePath, rowType); + latestRecordDataGeneratorCache.put( + tablePath, new BinaryRecordDataGenerator(cdcRowType)); + latestFieldConverterCache.put(tablePath, createFieldConverters(rowType)); + } + } + return inferSchemaChangeEvent; + } + + private RecordData convertFlussRowToCdcRecord( + TablePath tablePath, + InternalRow initialRow, + RowType initialRowType, + boolean isSchemaChangeEvent) { + RowType latestRowType = latestRowTypeCache.get(tablePath); + InternalRow row = initialRow; + + // A reader maybe subscribe multiple split, thus only inferred by the latest schema(also the + // widest) + if (isSchemaChangeEvent) { + org.apache.fluss.metadata.Schema latestSchema = + org.apache.fluss.metadata.Schema.newBuilder() + .fromRowType(latestRowType) + .build(); + org.apache.fluss.metadata.Schema currentSchema = + org.apache.fluss.metadata.Schema.newBuilder() + .fromRowType(initialRowType) + .build(); + row = ProjectedRow.from(currentSchema, latestSchema).replaceRow(initialRow); + } + + BinaryRecordDataGenerator generator = latestRecordDataGeneratorCache.get(tablePath); + FlussDeserializationConverter[] fieldConverters = latestFieldConverterCache.get(tablePath); + int fieldCount = latestRowType.getFieldCount(); + Object[] rowFields = new Object[fieldCount]; + for (int i = 0; i < fieldCount; i++) { + Object flussField = fieldConverters[i].getFieldOrNull(row); + rowFields[i] = fieldConverters[i].deserialize(flussField); + } + return generator.generate(rowFields); + } + + // ------------------------------------------------------------------------- + // Schema state restoration + // ------------------------------------------------------------------------- + + /** + * Restores the internal schema caches from a recovered split. This seeds the + * latestSchemaIdCache, latestRowTypeCache, and latestRecordDataGeneratorCache so that schema + * changes occurring after the last checkpoint can still be detected. + */ + @Override + public List<Event> restoreState(TablePath tablePath, int schemaId, RowType rowType) { + ensureCacheInitialized(); + // Multiple splits may read log with different schemaIds; only reserve the first one. + if (!latestSchemaIdCache.containsKey(tablePath)) { Review Comment: Could this split-initialization failure be propagated rather than returning after logging it? Failures while resolving startup offsets or creating snapshot/log splits currently leave the discovered buckets unassigned. In one-shot discovery mode the operation is never retried, so the job may remain RUNNING without producing data. A test that injects an offset-initialization failure and verifies that the source fails, rather than silently becoming idle, would help cover this behavior. ########## 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); + if (cachedSchemaId == null || schemaId > cachedSchemaId) { + if (cachedSchemaId == null) { + // First record for this table — emit CreateTableEvent with table keys. + events.add( + new CreateTableEvent( + tableId, + buildCdcSchema( + rowType, + record.getPrimaryKeyNames(), + record.getPartitionKeyNames()))); + } else { + // SchemaId changed — infer and emit schema change events + inferSchemaChangeEvent = true; + RowType oldRowType = latestRowTypeCache.get(tablePath); + events.addAll(inferSchemaChanges(tableId, tablePath, oldRowType, rowType)); + } + latestSchemaIdCache.put(tablePath, schemaId); + latestRowTypeCache.put(tablePath, rowType); + latestRecordDataGeneratorCache.put( + tablePath, new BinaryRecordDataGenerator(cdcRowType)); + latestFieldConverterCache.put(tablePath, createFieldConverters(rowType)); + } + } + return inferSchemaChangeEvent; + } + + private RecordData convertFlussRowToCdcRecord( + TablePath tablePath, + InternalRow initialRow, + RowType initialRowType, + boolean isSchemaChangeEvent) { + RowType latestRowType = latestRowTypeCache.get(tablePath); + InternalRow row = initialRow; + + // A reader maybe subscribe multiple split, thus only inferred by the latest schema(also the + // widest) + if (isSchemaChangeEvent) { Review Comment: Would it make sense to project the row whenever initialRowType differs from latestRowType, rather than only when the current record advances the cached schema ID? The Fluss MultiTableLogScanner uses dynamic schema resolution and preserves each record's original schema ID. With multiple buckets, records may therefore arrive in the order schema 1 → schema 2 → schema 1. After schema 2 updates the cache, the final schema 1 row is read using schema 2's field converters and currently fails with: ArrayIndexOutOfBoundsException: Index 2 out of bounds for length 2 Could we project every older-schema row to the cached latest schema? It would also be helpful to add a regression test covering schema 1 → 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]
