leonardBang commented on code in PR #4494: URL: https://github.com/apache/flink-cdc/pull/4494#discussion_r3853085978
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java: ########## @@ -0,0 +1,421 @@ +/* + * 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.reader; + +import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.WrapperFlussMetricRegistry; +import org.apache.flink.cdc.connectors.fluss.source.metrics.FlussSourceReaderMetrics; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSnapshotSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.connector.base.source.reader.RecordsBySplits; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.splitreader.SplitReader; +import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition; +import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.scanner.MultiTableRecord; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.client.table.scanner.batch.BatchScanner; +import org.apache.fluss.client.table.scanner.log.LogScanner; +import org.apache.fluss.client.table.scanner.log.MultiTableLogScanner; +import org.apache.fluss.client.table.scanner.log.MultiTableRecords; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataField; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.CloseableIterator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Queue; + +/** + * A {@link SplitReader} implementation for Fluss. It reads change log records from Fluss log + * scanners and wraps them as {@link FlussSourceRecord}s, which include the table context (table + * path and row type) needed for downstream deserialization. + * + * <p>For each table, a single Fluss {@link LogScanner} is shared across all bucket-level splits. + */ +public class FlussSplitReader implements SplitReader<FlussSourceRecord, FlussSplitBase> { + + private static final Logger LOG = LoggerFactory.getLogger(FlussSplitReader.class); + private static final Duration POLL_TIMEOUT = Duration.ofMillis(100); + private static final Duration BATCH_POLL_TIMEOUT = Duration.ofMillis(10000L); + + private final Configuration flussConfig; + private final WrapperFlussMetricRegistry metricRegistry; + private final FlussSourceReaderMetrics sourceReaderMetrics; + private Connection connection; + private final Map<TablePath, Table> tables; + private final Map<TablePath, RowType> tableRowTypes; + private final Map<TablePath, List<String>> tablePrimaryKeyNames; + private final Map<TablePath, List<String>> tablePartitionKeyNames; + private final Map<TableBucket, FlussSplitBase> bucketToSplit; + + // Bounded (snapshot) split reading + private final Queue<FlussSplitBase> boundedSplits; + @Nullable private FlussSplitBase currentBoundedSplit; + @Nullable private BatchScanner currentBatchScanner; + @Nullable private Integer currentBatchSchemaId; + @Nullable private MultiTableLogScanner currentLogScanner; + private long snapshotRecordsToSkip; + private long currentReadRecordsCount; + + public FlussSplitReader( + Configuration flussConfig, + WrapperFlussMetricRegistry metricRegistry, + FlussSourceReaderMetrics sourceReaderMetrics) { + this.flussConfig = flussConfig; + this.metricRegistry = metricRegistry; + this.sourceReaderMetrics = sourceReaderMetrics; + this.tables = new HashMap<>(); + this.tableRowTypes = new HashMap<>(); + this.tablePrimaryKeyNames = new HashMap<>(); + this.tablePartitionKeyNames = new HashMap<>(); + this.bucketToSplit = new HashMap<>(); + this.boundedSplits = new ArrayDeque<>(); + } + + @Override + public RecordsWithSplitIds<FlussSourceRecord> fetch() throws IOException { + RecordsBySplits.Builder<FlussSourceRecord> builder = new RecordsBySplits.Builder<>(); + + // Priority: read bounded (snapshot) splits first, then log + checkSnapshotSplitOrStartNext(); + if (currentBatchScanner != null) { + fetchSnapshotRecords(builder); + return builder.build(); + } + + // Read from log scanners + long fetchTimestamp = System.currentTimeMillis(); + long maxRecordTimestamp = -1; + + MultiTableLogScanner scanner = getOrCreateTableLogScanner(); + MultiTableRecords scanRecords = scanner.poll(POLL_TIMEOUT); + if (scanRecords != null && !scanRecords.isEmpty()) { + for (TablePath tablePath : scanRecords.tablePaths()) { + for (TableBucket bucket : scanRecords.buckets(tablePath)) { + for (MultiTableRecord record : scanRecords.records(tablePath, bucket)) { + FlussSplitBase split = bucketToSplit.get(bucket); + if (split == null) { + LOG.warn("Received records for unknown bucket {}, skipping", bucket); + continue; + } + builder.add( + split.splitId(), + new FlussSourceRecord( + record, getPartitionKeyNames(record.getTablePath()))); + + maxRecordTimestamp = Math.max(maxRecordTimestamp, record.timestamp()); + } + } + } + } + + // Report event time lag + if (maxRecordTimestamp > 0) { + sourceReaderMetrics.reportRecordEventTime(fetchTimestamp - maxRecordTimestamp); Review Comment: Could we calculate the fetch event-time lag using the current time after `poll()` completes? `fetchTimestamp` is captured before the blocking poll. A record appended while the poll is waiting may therefore have a timestamp later than `fetchTimestamp`, causing `currentFetchEventTimeLag` to report a negative value. It may also be helpful to clamp the result to zero to tolerate clock skew and add a test covering a record timestamp later than the poll start time. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSplitReader.java: ########## @@ -0,0 +1,421 @@ +/* + * 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.reader; + +import org.apache.flink.cdc.connectors.fluss.sink.v2.metrics.WrapperFlussMetricRegistry; +import org.apache.flink.cdc.connectors.fluss.source.metrics.FlussSourceReaderMetrics; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussHybridSnapshotLogSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSnapshotSplit; +import org.apache.flink.cdc.connectors.fluss.source.split.FlussSplitBase; +import org.apache.flink.connector.base.source.reader.RecordsBySplits; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.splitreader.SplitReader; +import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition; +import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.scanner.MultiTableRecord; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.client.table.scanner.batch.BatchScanner; +import org.apache.fluss.client.table.scanner.log.LogScanner; +import org.apache.fluss.client.table.scanner.log.MultiTableLogScanner; +import org.apache.fluss.client.table.scanner.log.MultiTableRecords; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataField; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.CloseableIterator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Queue; + +/** + * A {@link SplitReader} implementation for Fluss. It reads change log records from Fluss log + * scanners and wraps them as {@link FlussSourceRecord}s, which include the table context (table + * path and row type) needed for downstream deserialization. + * + * <p>For each table, a single Fluss {@link LogScanner} is shared across all bucket-level splits. + */ +public class FlussSplitReader implements SplitReader<FlussSourceRecord, FlussSplitBase> { + + private static final Logger LOG = LoggerFactory.getLogger(FlussSplitReader.class); + private static final Duration POLL_TIMEOUT = Duration.ofMillis(100); + private static final Duration BATCH_POLL_TIMEOUT = Duration.ofMillis(10000L); + + private final Configuration flussConfig; + private final WrapperFlussMetricRegistry metricRegistry; + private final FlussSourceReaderMetrics sourceReaderMetrics; + private Connection connection; + private final Map<TablePath, Table> tables; + private final Map<TablePath, RowType> tableRowTypes; + private final Map<TablePath, List<String>> tablePrimaryKeyNames; + private final Map<TablePath, List<String>> tablePartitionKeyNames; + private final Map<TableBucket, FlussSplitBase> bucketToSplit; + + // Bounded (snapshot) split reading + private final Queue<FlussSplitBase> boundedSplits; + @Nullable private FlussSplitBase currentBoundedSplit; + @Nullable private BatchScanner currentBatchScanner; + @Nullable private Integer currentBatchSchemaId; + @Nullable private MultiTableLogScanner currentLogScanner; + private long snapshotRecordsToSkip; + private long currentReadRecordsCount; + + public FlussSplitReader( + Configuration flussConfig, + WrapperFlussMetricRegistry metricRegistry, + FlussSourceReaderMetrics sourceReaderMetrics) { + this.flussConfig = flussConfig; + this.metricRegistry = metricRegistry; + this.sourceReaderMetrics = sourceReaderMetrics; + this.tables = new HashMap<>(); + this.tableRowTypes = new HashMap<>(); + this.tablePrimaryKeyNames = new HashMap<>(); + this.tablePartitionKeyNames = new HashMap<>(); + this.bucketToSplit = new HashMap<>(); + this.boundedSplits = new ArrayDeque<>(); + } + + @Override + public RecordsWithSplitIds<FlussSourceRecord> fetch() throws IOException { + RecordsBySplits.Builder<FlussSourceRecord> builder = new RecordsBySplits.Builder<>(); + + // Priority: read bounded (snapshot) splits first, then log + checkSnapshotSplitOrStartNext(); + if (currentBatchScanner != null) { + fetchSnapshotRecords(builder); + return builder.build(); + } + + // Read from log scanners + long fetchTimestamp = System.currentTimeMillis(); + long maxRecordTimestamp = -1; + + MultiTableLogScanner scanner = getOrCreateTableLogScanner(); + MultiTableRecords scanRecords = scanner.poll(POLL_TIMEOUT); + if (scanRecords != null && !scanRecords.isEmpty()) { + for (TablePath tablePath : scanRecords.tablePaths()) { + for (TableBucket bucket : scanRecords.buckets(tablePath)) { + for (MultiTableRecord record : scanRecords.records(tablePath, bucket)) { + FlussSplitBase split = bucketToSplit.get(bucket); + if (split == null) { + LOG.warn("Received records for unknown bucket {}, skipping", bucket); + continue; + } + builder.add( + split.splitId(), + new FlussSourceRecord( + record, getPartitionKeyNames(record.getTablePath()))); + + maxRecordTimestamp = Math.max(maxRecordTimestamp, record.timestamp()); + } + } + } + } + + // Report event time lag + if (maxRecordTimestamp > 0) { + sourceReaderMetrics.reportRecordEventTime(fetchTimestamp - maxRecordTimestamp); + } + + return builder.build(); + } + + @Override + public void handleSplitsChanges(SplitsChange<FlussSplitBase> splitsChanges) { + if (!(splitsChanges instanceof SplitsAddition)) { + throw new UnsupportedOperationException( + String.format( + "The SplitChange type of %s is not supported.", + splitsChanges.getClass())); + } + + if (connection == null) { + connection = ConnectionFactory.createConnection(flussConfig, metricRegistry); + } + + for (FlussSplitBase split : splitsChanges.splits()) { + if (!split.isHybridSnapshotLogSplit() && !split.isLogSplit()) { + LOG.warn("Unsupported split type: {}, skipping", split.getClass().getSimpleName()); + continue; + } + Table table = getOrCreateTable(split.getTablePath()); + validateTableId(split, table.getTableInfo().getTableId()); + if (split.isHybridSnapshotLogSplit()) { + FlussHybridSnapshotLogSplit hybrid = split.asHybridSnapshotLogSplit(); + // If snapshot is not finished, add to pending bounded splits + if (!hybrid.isSnapshotFinished()) { + boundedSplits.add(split); + } + // Still need to subscribe log for after snapshot reading + subscribeLog(split, hybrid.getLogStartingOffset()); + } else { + subscribeLog(split, split.asLogSplit().getStartingOffset()); + } + } + } + + // ------------------------------------------------------------------------- + // Bounded (snapshot) split reading + // ------------------------------------------------------------------------- + + /** If no bounded split is being read, poll the next one from the queue and start reading. */ + private void checkSnapshotSplitOrStartNext() { + if (currentBatchScanner != null) { + return; + } + + FlussSplitBase nextSplit = boundedSplits.poll(); + if (nextSplit == null) { + return; + } + + currentBoundedSplit = nextSplit; + FlussSnapshotSplit snapshotSplit = nextSplit.asSnapshotSplit(); + Table table = getOrCreateTable(nextSplit.getTablePath()); + currentBatchSchemaId = table.getTableInfo().getSchemaId(); + currentBatchScanner = + table.newScan() + .createBatchScanner( + snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId()); + snapshotRecordsToSkip = snapshotSplit.getRecordsToSkip(); + currentReadRecordsCount = 0; + LOG.info("Started reading snapshot for split {}", nextSplit.splitId()); + } + + /** + * Reads a batch of snapshot records. On recovery, skips records that have already been + * processed. Each emitted record carries its cumulative {@code readRecordsCount}. + */ + private void fetchSnapshotRecords(RecordsBySplits.Builder<FlussSourceRecord> builder) + throws IOException { + assert currentBoundedSplit != null; + assert currentBatchSchemaId != null; + assert currentBatchScanner != null; + TablePath tablePath = currentBoundedSplit.getTablePath(); + RowType rowType = getRowType(tablePath); + + CloseableIterator<InternalRow> batch = currentBatchScanner.pollBatch(BATCH_POLL_TIMEOUT); + if (batch == null) { + // Snapshot fully read + finishCurrentBoundedSplit(builder); + return; + } + + try { + while (batch.hasNext()) { + InternalRow row = batch.next(); + currentReadRecordsCount++; + if (snapshotRecordsToSkip > 0) { + snapshotRecordsToSkip--; + continue; + } + ScanRecord scanRecord = + new ScanRecord( + currentBoundedSplit.getTableBucket().getTableId(), + currentBatchSchemaId, + -1L, + -1L, + ChangeType.INSERT, + row, + // TODO: Calculate the actual record size in bytes. + 1); + builder.add( + currentBoundedSplit.splitId(), + new FlussSourceRecord( + scanRecord, + tablePath, + rowType, + currentReadRecordsCount, + getPrimaryKeyNames(tablePath), + getPartitionKeyNames(tablePath))); + } + } finally { + batch.close(); + } + } + + /** + * Called when the current bounded split's snapshot is fully read. For hybrid splits, the split + * is NOT marked as finished since log reading continues. For pure snapshot splits, the split is + * marked as finished. + */ + private void finishCurrentBoundedSplit(RecordsBySplits.Builder<FlussSourceRecord> builder) + throws IOException { + if (currentBoundedSplit.isHybridSnapshotLogSplit()) { + // Hybrid split: snapshot done, log reading continues — do NOT finish the split + LOG.info("Snapshot phase finished for hybrid split {}", currentBoundedSplit.splitId()); + } else { + // Pure snapshot split: mark as finished + builder.addFinishedSplit(currentBoundedSplit.splitId()); + LOG.info("Snapshot split {} finished", currentBoundedSplit.splitId()); + } + closeCurrentBoundedSplit(); + } + + private void closeCurrentBoundedSplit() throws IOException { + try { + if (currentBatchScanner != null) { + currentBatchScanner.close(); + } + } catch (Exception e) { + throw new IOException("Failed to close batch scanner", e); + } + + // todo: 可以封装为一个对象 Review Comment: Could we remove this TODO or rewrite it in English? Comments in the project should be written in English. For example: `// TODO: Encapsulate the snapshot reader state in a dedicated object.` ########## 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,503 @@ +/* + * 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(); + int schemaId = record.getScanRecord().getSchemaId(); + + inferSchemaChangeEvents(events, record, tablePath, tableId, schemaId); + InternalRow row = record.getScanRecord().getRow(); + ChangeType changeType = record.getScanRecord().getChangeType(); + + switch (changeType) { + case APPEND_ONLY: + case INSERT: + { + RecordData after = + convertFlussRowToCdcRecord(tablePath, row, rowType, schemaId); + 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, schemaId); + events.add(DataChangeEvent.replaceEvent(tableId, after)); + break; + } + case DELETE: + { + RecordData before = + convertFlussRowToCdcRecord(tablePath, row, rowType, schemaId); + events.add(DataChangeEvent.deleteEvent(tableId, before)); + break; + } + default: + throw new IllegalArgumentException("Unsupported change type: " + changeType); + } + return events; + } + + private void inferSchemaChangeEvents( + List<Event> events, + FlussSourceRecord record, + TablePath tablePath, + TableId tableId, + int schemaId) { + // Detect schema changes using the schema ID carried by each source record. + RowType rowType = record.getRowType(); + 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 + RowType oldRowType = latestRowTypeCache.get(tablePath); + events.addAll(inferSchemaChanges(tableId, tablePath, oldRowType, rowType)); + } + latestSchemaIdCache.put(tablePath, schemaId); + latestRowTypeCache.put(tablePath, rowType); + org.apache.flink.cdc.common.types.RowType cdcRowType = + (org.apache.flink.cdc.common.types.RowType) + FlussConversions.toCdcType(rowType); + latestRecordDataGeneratorCache.put( + tablePath, new BinaryRecordDataGenerator(cdcRowType)); + latestFieldConverterCache.put(tablePath, createFieldConverters(rowType)); + } + } + } + + private RecordData convertFlussRowToCdcRecord( + TablePath tablePath, + InternalRow initialRow, + RowType initialRowType, + int initialSchemaId) { + RowType latestRowType = latestRowTypeCache.get(tablePath); + InternalRow row = initialRow; + + // Records emitted by FlussSplitReader always carry a valid schema ID. Compare IDs to avoid + // traversing the RowType on every record. + int latestSchemaId = latestSchemaIdCache.get(tablePath); + if (initialSchemaId != latestSchemaId) { + 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. Review Comment: Could we retain the restored schema with the highest schema ID here instead of whichever split is initialized first? Different splits of the same table may checkpoint different schema IDs, and `initializedState()` applies them independently during recovery. If the older split is processed first, `restoredCreateTableRowTypeCache` is initialized with an outdated schema and the first post-recovery record emits an outdated `CreateTableEvent`. Could we replace the cached schema whenever `schemaId` is greater than the currently restored schema ID? It would also be helpful to cover both restoration 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]
