diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java index 93261e8430a..a4429753482 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java @@ -25,22 +25,30 @@ import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.predicate.Predicate; import org.apache.fluss.types.RowType; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceProvider; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import javax.annotation.Nullable; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.Map; /** A Flink table source for the $binlog virtual table. */ -public class BinlogFlinkTableSource implements ScanTableSource { +public class BinlogFlinkTableSource + implements ScanTableSource, SupportsProjectionPushDown, SupportsFilterPushDown { private final TablePath tablePath; private final Configuration flussConfig; @@ -55,12 +63,12 @@ public class BinlogFlinkTableSource implements ScanTableSource { private final int splitPerAssignmentBatchSize; private final Map tableOptions; - // Projection pushdown - @Nullable private int[] projectedFields; + // Projection pushdown. Top-level projection over the binlog row [_change_type, _log_offset, + // _commit_timestamp, before, after]; the underlying data scan stays full because before/after + // are whole nested ROWs (nested pruning is out of scope). + @Nullable private int[] projectedTopLevel; private LogicalType producedDataType; - @Nullable private Predicate partitionFilters; - public BinlogFlinkTableSource( TablePath tablePath, Configuration flussConfig, @@ -116,11 +124,9 @@ public ChangelogMode getChangelogMode() { @Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { - // Create the Fluss row type for the data columns (the original table columns) + // Create the Fluss row type for the data columns (the original table columns). The data + // scan always stays full because the projected before/after columns are whole nested ROWs. RowType flussRowType = FlinkConversions.toFlussRowType(dataColumnsType); - if (projectedFields != null) { - flussRowType = flussRowType.project(projectedFields); - } // Determine the offsets initializer based on startup mode OffsetsInitializer offsetsInitializer; @@ -150,14 +156,18 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { false, isPartitioned, flussRowType, - projectedFields, + null, null, offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new BinlogDeserializationSchema(), + new BinlogDeserializationSchema( + projectedTopLevel, + (org.apache.flink.table.types.logical.RowType) producedDataType), streaming, - partitionFilters, + // $binlog data/partition columns are nested inside before/after ROWs, so no + // top-level partition filter is pushable; always scan without one. + null, LeaseContext.DEFAULT); return SourceProvider.of(source); @@ -177,8 +187,7 @@ public DynamicTableSource copy() { splitPerAssignmentBatchSize, tableOptions); copy.producedDataType = producedDataType; - copy.projectedFields = projectedFields; - copy.partitionFilters = partitionFilters; + copy.projectedTopLevel = projectedTopLevel; return copy; } @@ -187,6 +196,39 @@ public String asSummaryString() { return "FlussBinlogTableSource"; } - // TODO: Implement projection pushdown handling for nested before/after columns - // TODO: Implement filter pushdown + @Override + public boolean supportsNestedProjection() { + return false; + } + + @Override + public void applyProjection(int[][] projectedFields, DataType producedDataType) { + // Top-level projection over [_change_type, _log_offset, _commit_timestamp, before, after]. + // The data scan stays full (before/after are whole nested ROWs); the deserialization schema + // emits only the projected top-level columns. + this.projectedTopLevel = + Arrays.stream(projectedFields).mapToInt(value -> value[0]).toArray(); + this.producedDataType = producedDataType.getLogicalType(); + } + + @Override + public Result applyFilters(List filters) { + // The $binlog data and partition columns are nested inside the before/after ROW columns, + // and the leading columns are non-pushable metadata (_change_type, _log_offset, + // _commit_timestamp). No top-level filter is convertible to a Fluss predicate, so nothing + // is pushed down; all filters are returned to Flink. Implemented for interface parity with + // the normal table. + return Result.of(Collections.emptyList(), filters); + } + + @VisibleForTesting + @Nullable + int[] getProjectedTopLevel() { + return projectedTopLevel; + } + + @VisibleForTesting + LogicalType getProducedDataType() { + return producedDataType; + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java index 4304b271b28..1723057541f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java @@ -19,34 +19,52 @@ import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; import org.apache.fluss.flink.FlinkConnectorOptions; import org.apache.fluss.flink.source.deserializer.ChangelogDeserializationSchema; import org.apache.fluss.flink.source.reader.LeaseContext; import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; import org.apache.fluss.flink.utils.FlinkConversions; +import org.apache.fluss.flink.utils.PushdownUtils; +import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.predicate.PartitionPredicateVisitor; import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.predicate.PredicateBuilder; +import org.apache.fluss.predicate.PredicateVisitor; import org.apache.fluss.types.RowType; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceProvider; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; +import static org.apache.fluss.flink.utils.PredicateConverter.convertToFlussPredicate; + /** A Flink table source for the $changelog virtual table. */ -public class ChangelogFlinkTableSource implements ScanTableSource { +public class ChangelogFlinkTableSource + implements ScanTableSource, SupportsProjectionPushDown, SupportsFilterPushDown { private final TablePath tablePath; private final Configuration flussConfig; @@ -64,8 +82,17 @@ public class ChangelogFlinkTableSource implements ScanTableSource { // Projection pushdown @Nullable private int[] projectedFields; private LogicalType producedDataType; + // Data-column indices actually scanned from Fluss (derived from the virtual projection). + @Nullable private int[] dataProjection; + // Projection over the base changelog row [metadata columns + scanned data columns]. + @Nullable private int[] baseRowProjection; @Nullable private Predicate partitionFilters; + // Filter over the scanned data columns pushed to the Fluss log scan (statistics-based). + @Nullable private Predicate logRecordBatchFilter; + + // Number of leading changelog metadata columns: _change_type, _log_offset, _commit_timestamp. + private static final int METADATA_COLUMN_COUNT = 3; private static final Set METADATA_COLUMN_NAMES = new HashSet<>( @@ -147,10 +174,8 @@ public ChangelogMode getChangelogMode() { public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { // Create the Fluss row type for the data columns (without metadata) RowType flussRowType = FlinkConversions.toFlussRowType(dataColumnsType); - if (projectedFields != null) { - // Adjust projection to account for metadata columns - // TODO: Handle projection properly with metadata columns - flussRowType = flussRowType.project(projectedFields); + if (dataProjection != null) { + flussRowType = flussRowType.project(dataProjection); } // to capture all change types (+I, -U, +U, -D). // FULL mode reads snapshot first (no change types), so we use EARLIEST for log-only @@ -187,12 +212,14 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { false, isPartitioned(), flussRowType, - projectedFields, - null, + dataProjection, + logRecordBatchFilter, offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new ChangelogDeserializationSchema(), + new ChangelogDeserializationSchema( + baseRowProjection, + (org.apache.flink.table.types.logical.RowType) producedDataType), streaming, partitionFilters, LeaseContext.DEFAULT); // Lake source not supported @@ -215,7 +242,10 @@ public DynamicTableSource copy() { tableOptions); copy.producedDataType = producedDataType; copy.projectedFields = projectedFields; + copy.dataProjection = dataProjection; + copy.baseRowProjection = baseRowProjection; copy.partitionFilters = partitionFilters; + copy.logRecordBatchFilter = logRecordBatchFilter; return copy; } @@ -224,10 +254,177 @@ public String asSummaryString() { return "FlussChangelogTableSource"; } - // TODO: Implement projection pushdown handling for metadata columns - // TODO: Implement filter pushdown + @Override + public boolean supportsNestedProjection() { + return false; + } + + @Override + public void applyProjection(int[][] projectedFields, DataType producedDataType) { + this.projectedFields = Arrays.stream(projectedFields).mapToInt(value -> value[0]).toArray(); + this.producedDataType = producedDataType.getLogicalType(); + + // Derive the data-column indices actually scanned from Fluss (first-appearance order), + // and the projection over the base changelog row [metadata columns + scanned data columns]. + Map dataColumnPositions = new LinkedHashMap<>(); + for (int virtualIndex : this.projectedFields) { + if (virtualIndex >= METADATA_COLUMN_COUNT) { + dataColumnPositions.putIfAbsent( + virtualIndex - METADATA_COLUMN_COUNT, dataColumnPositions.size()); + } + } + + // When no data column is projected (metadata-only projection, e.g. SELECT _change_type), + // normalize the scan projection to null so Fluss reads the full data row. A zero-column + // scan produces no records, so the metadata columns must be emitted from a full data scan + // and selected later during deserialization via baseRowProjection. + this.dataProjection = + dataColumnPositions.isEmpty() + ? null + : dataColumnPositions.keySet().stream() + .mapToInt(Integer::intValue) + .toArray(); + + int[] base = new int[this.projectedFields.length]; + for (int i = 0; i < this.projectedFields.length; i++) { + int virtualIndex = this.projectedFields[i]; + base[i] = + virtualIndex < METADATA_COLUMN_COUNT + ? virtualIndex + : METADATA_COLUMN_COUNT + + dataColumnPositions.get(virtualIndex - METADATA_COLUMN_COUNT); + } + this.baseRowProjection = base; + } + + @Override + public Result applyFilters(List filters) { + List acceptedFilters = new ArrayList<>(); + List remainingFilters = new ArrayList<>(); + + RowType dataFlussRowType = FlinkConversions.toFlussRowType(dataColumnsType); + + if (isPartitioned()) { + // apply partition filter pushdown + List converted = new ArrayList<>(); + RowType partitionRowType = dataFlussRowType.project(partitionKeyIndexes); + PredicateVisitor checksOnlyPartitionKeys = + new PartitionPredicateVisitor(partitionRowType.getFieldNames()); + + for (ResolvedExpression filter : filters) { + Optional predicateOptional = + convertToFlussPredicate(partitionRowType, filter); + if (predicateOptional.isPresent()) { + Predicate p = predicateOptional.get(); + // partition pushdown can only guarantee to filter out partitions matching the + // predicate, but can't guarantee to filter out all data matching a + // non-partition filter in the partition + if (!p.visit(checksOnlyPartitionKeys)) { + remainingFilters.add(filter); + } else { + acceptedFilters.add(filter); + converted.add(p); + } + } else { + remainingFilters.add(filter); + } + } + partitionFilters = converted.isEmpty() ? null : PredicateBuilder.and(converted); + } + + if (acceptedFilters.isEmpty() && remainingFilters.isEmpty()) { + remainingFilters.addAll(filters); + } + + // Data-column log filter pushdown (statistics-based). Metadata columns (_change_type, + // _log_offset, _commit_timestamp) are resolved by name against the data columns and + // therefore naturally fall out as non-pushable. + Result recordBatchResult = pushdownRecordBatchFilter(remainingFilters, dataFlussRowType); + acceptedFilters.addAll(recordBatchResult.getAcceptedFilters()); + + // We cannot determine whether this source will ultimately be used as a scan source. Always + // return all original filters as remaining so Flink applies them as a safety net. + return Result.of(acceptedFilters, filters); + } + + private Result pushdownRecordBatchFilter( + List filters, RowType dataFlussRowType) { + TableConfig tableConfig = new TableConfig(Configuration.fromMap(tableOptions)); + // Log filter pushdown is only supported for the ARROW log format. For INDEXED/COMPACTED + // formats, TableScan#createLogScanner rejects a record-batch filter at runtime, so we skip + // pushdown here and let Flink apply the filters as a post-scan safety net. + if (tableConfig.getLogFormat() != LogFormat.ARROW) { + this.logRecordBatchFilter = null; + return Result.of(Collections.emptyList(), filters); + } + + Set availableStatsColumns = + PushdownUtils.computeAvailableStatsColumns(dataFlussRowType, tableConfig); + + List pushdownPredicates = new ArrayList<>(); + List acceptedFilters = new ArrayList<>(); + List remainingFilters = new ArrayList<>(); + + for (ResolvedExpression filter : filters) { + Optional predicateOpt = convertToFlussPredicate(dataFlussRowType, filter); + if (predicateOpt.isPresent() + && PushdownUtils.canPredicateUseStatistics( + predicateOpt.get(), dataFlussRowType, availableStatsColumns)) { + pushdownPredicates.add(predicateOpt.get()); + acceptedFilters.add(filter); + } + // All filters are kept as remaining so that Flink can still verify the results + // after server-side filtering (safety net). + remainingFilters.add(filter); + } + + if (pushdownPredicates.isEmpty()) { + this.logRecordBatchFilter = null; + } else { + this.logRecordBatchFilter = + pushdownPredicates.size() == 1 + ? pushdownPredicates.get(0) + : PredicateBuilder.and(pushdownPredicates); + } + return Result.of(acceptedFilters, remainingFilters); + } private boolean isPartitioned() { return partitionKeyIndexes.length > 0; } + + @VisibleForTesting + @Nullable + int[] getProjectedFields() { + return projectedFields; + } + + @VisibleForTesting + @Nullable + int[] getDataProjection() { + return dataProjection; + } + + @VisibleForTesting + @Nullable + int[] getBaseRowProjection() { + return baseRowProjection; + } + + @VisibleForTesting + LogicalType getProducedDataType() { + return producedDataType; + } + + @VisibleForTesting + @Nullable + Predicate getLogRecordBatchFilter() { + return logRecordBatchFilter; + } + + @VisibleForTesting + @Nullable + Predicate getPartitionFilters() { + return partitionFilters; + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java index 24f3d94d670..162993d8f6c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java @@ -21,7 +21,6 @@ import org.apache.fluss.client.table.getter.PartitionGetter; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; -import org.apache.fluss.config.StatisticsColumnsConfig; import org.apache.fluss.config.TableConfig; import org.apache.fluss.flink.FlinkConnectorOptions; import org.apache.fluss.flink.row.FlinkAsFlussRow; @@ -42,12 +41,10 @@ import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.predicate.CompoundPredicate; import org.apache.fluss.predicate.PartitionPredicateVisitor; import org.apache.fluss.predicate.Predicate; import org.apache.fluss.predicate.PredicateBuilder; import org.apache.fluss.predicate.PredicateVisitor; -import org.apache.fluss.types.DataTypeChecks; import org.apache.fluss.types.RowType; import org.apache.flink.annotation.VisibleForTesting; @@ -266,7 +263,8 @@ public FlinkTableSource( // Pre-compute available statistics columns to avoid repeated calculation RowType flussRowType = FlinkConversions.toFlussRowType(tableOutputType); - this.availableStatsColumns = computeAvailableStatsColumns(flussRowType); + this.availableStatsColumns = + PushdownUtils.computeAvailableStatsColumns(flussRowType, tableConfig); } @Override @@ -678,7 +676,8 @@ private Result pushdownRecordBatchFilter(List filters) { Predicate predicate = predicateOpt.get(); LOG.trace("Converted filter to predicate: {}", predicate); // Check if predicate can benefit from statistics - if (canPredicateUseStatistics(predicate, flussRowType, availableStatsColumns)) { + if (PushdownUtils.canPredicateUseStatistics( + predicate, flussRowType, availableStatsColumns)) { pushdownPredicates.add(predicate); acceptedFilters.add(filter); } @@ -731,89 +730,6 @@ private void pushdownLakeFilters( } } - /** - * Checks if a predicate can benefit from statistics based on the available statistics columns. - * - * @param predicate the predicate to check - * @param rowType the row type - * @param availableStatsColumns the columns that have statistics available - * @return true if the predicate can use statistics - */ - private boolean canPredicateUseStatistics( - Predicate predicate, RowType rowType, Set availableStatsColumns) { - - class StatisticsUsageVisitor implements PredicateVisitor { - @Override - public Boolean visit(org.apache.fluss.predicate.LeafPredicate leaf) { - // Check if the field referenced by this predicate has statistics available - String fieldName = rowType.getFieldNames().get(leaf.index()); - // Check if statistics are available for this column - return availableStatsColumns.contains(fieldName); - } - - @Override - public Boolean visit(CompoundPredicate compound) { - // For compound predicates, all children must be able to use statistics - for (Predicate child : compound.children()) { - if (!child.visit(this)) { - return false; - } - } - return true; - } - } - - return predicate.visit(new StatisticsUsageVisitor()); - } - - /** - * Computes the available statistics columns based on table configuration. This method is called - * once during construction to pre-compute the result. - * - * @param flussRowType the row type - * @return set of column names that have statistics available - */ - private Set computeAvailableStatsColumns(RowType flussRowType) { - StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns(); - - if (!statsConfig.isEnabled()) { - LOG.debug("Statistics collection is disabled for the table"); - return Collections.emptySet(); - } - - Set columns = new HashSet<>(); - if (statsConfig.getMode() == StatisticsColumnsConfig.Mode.ALL) { - // Collect all columns with supported statistics types - for (int i = 0; i < flussRowType.getFieldCount(); i++) { - org.apache.fluss.types.DataType fieldType = flussRowType.getTypeAt(i); - if (DataTypeChecks.isSupportedStatisticsType(fieldType)) { - columns.add(flussRowType.getFieldNames().get(i)); - } - } - } else { - // Use user-specified columns (validate they exist and have supported types) - for (String columnName : statsConfig.getColumns()) { - int columnIndex = flussRowType.getFieldNames().indexOf(columnName); - if (columnIndex >= 0) { - org.apache.fluss.types.DataType fieldType = flussRowType.getTypeAt(columnIndex); - if (DataTypeChecks.isSupportedStatisticsType(fieldType)) { - columns.add(columnName); - } else { - LOG.trace( - "Configured statistics column '{}' has unsupported type and will be ignored", - columnName); - } - } else { - LOG.trace( - "Configured statistics column '{}' does not exist in table schema", - columnName); - } - } - } - - return columns; - } - @Override public RowLevelModificationScanContext applyRowLevelModificationScan( RowLevelModificationType rowLevelModificationType, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java index 1114febb9c9..923417251f2 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java @@ -27,8 +27,6 @@ import javax.annotation.Nullable; -import static org.apache.fluss.flink.utils.FlinkConversions.toFlinkRowType; - /** * A deserialization schema that converts {@link LogRecord} objects to Flink's {@link RowData} * format with nested before/after row structure for the $binlog virtual table. @@ -46,14 +44,33 @@ public class BinlogDeserializationSchema implements FlussDeserializationSchema getProducedType(RowType rowSchema) { - // Build the output type with nested before/after ROW columns - org.apache.flink.table.types.logical.RowType outputType = - BinlogRowConverter.buildBinlogRowType(toFlinkRowType(rowSchema)); - return InternalTypeInfo.of(outputType); + // The produced type has already been computed by the table source (handed over by the + // Flink planner via applyProjection); no need to derive it again here. + return InternalTypeInfo.of(producedRowType); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java index c81141d345c..98ff62c5822 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java @@ -25,7 +25,7 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; -import static org.apache.fluss.flink.utils.FlinkConversions.toFlinkRowType; +import javax.annotation.Nullable; /** * A deserialization schema that converts {@link LogRecord} objects to Flink's {@link RowData} @@ -39,14 +39,33 @@ public class ChangelogDeserializationSchema implements FlussDeserializationSchem */ private transient ChangelogRowConverter converter; - /** Creates a new ChangelogDeserializationSchema. */ - public ChangelogDeserializationSchema() {} + /** + * Optional projection over the base changelog row {@code [_change_type, _log_offset, + * _commit_timestamp, ]}, or {@code null} for no projection. + */ + @Nullable private final int[] baseRowProjection; + + /** The produced row type, already computed by the table source honoring the projection. */ + private final org.apache.flink.table.types.logical.RowType producedRowType; + + /** + * Creates a new ChangelogDeserializationSchema. + * + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + * @param producedRowType the produced row type computed by the table source + */ + public ChangelogDeserializationSchema( + @Nullable int[] baseRowProjection, + org.apache.flink.table.types.logical.RowType producedRowType) { + this.baseRowProjection = baseRowProjection; + this.producedRowType = producedRowType; + } /** Initializes the deserialization schema. */ @Override public void open(InitializationContext context) throws Exception { if (converter == null) { - this.converter = new ChangelogRowConverter(context.getRowSchema()); + this.converter = new ChangelogRowConverter(context.getRowSchema(), baseRowProjection); } } @@ -67,9 +86,8 @@ public RowData deserialize(LogRecord record) throws Exception { */ @Override public TypeInformation getProducedType(RowType rowSchema) { - // Build the output type with metadata columns - org.apache.flink.table.types.logical.RowType outputType = - ChangelogRowConverter.buildChangelogRowType(toFlinkRowType(rowSchema)); - return InternalTypeInfo.of(outputType); + // The produced type has already been computed by the table source (handed over by the + // Flink planner via applyProjection); no need to derive it again here. + return InternalTypeInfo.of(producedRowType); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java index 3a9f2c94116..c47fdf7dd2e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java @@ -26,6 +26,7 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.data.utils.ProjectedRowData; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.VarCharType; @@ -45,16 +46,36 @@ public class BinlogRowConverter implements RecordToFlinkRowConverter { private final FlussRowToFlinkRowConverter baseConverter; private final org.apache.flink.table.types.logical.RowType producedType; + /** + * Optional top-level projection over the binlog row {@code [_change_type, _log_offset, + * _commit_timestamp, before, after]}. When non-null, the converter emits exactly those columns + * in that order; {@code null} means the full binlog row is emitted. Nested pruning inside + * before/after is not supported, so the underlying data scan always reads all columns. + */ + @Nullable private final int[] projectedTopLevel; + /** * Buffer for the UPDATE_BEFORE (-U) record pending merge with the next UPDATE_AFTER (+U) * record. Null when no update is in progress. */ @Nullable private LogRecord pendingUpdateBefore; - /** Creates a new BinlogRowConverter. */ + /** Creates a new BinlogRowConverter without projection. */ public BinlogRowConverter(RowType rowType) { + this(rowType, null); + } + + /** + * Creates a new BinlogRowConverter. + * + * @param rowType the data columns scanned from Fluss (always the full set) + * @param projectedTopLevel top-level projection over the binlog row, or {@code null} for none + */ + public BinlogRowConverter(RowType rowType, @Nullable int[] projectedTopLevel) { this.baseConverter = new FlussRowToFlinkRowConverter(rowType); - this.producedType = buildBinlogRowType(FlinkConversions.toFlinkRowType(rowType)); + this.projectedTopLevel = projectedTopLevel; + this.producedType = + buildProducedRowType(FlinkConversions.toFlinkRowType(rowType), projectedTopLevel); } /** Converts a LogRecord to a binlog RowData with nested before/after structure. */ @@ -137,7 +158,10 @@ private RowData buildBinlogRow( row.setField(3, before); row.setField(4, after); row.setRowKind(RowKind.INSERT); - return row; + if (projectedTopLevel == null) { + return row; + } + return ProjectedRowData.from(projectedTopLevel).replaceRow(row); } /** @@ -171,4 +195,27 @@ public static org.apache.flink.table.types.logical.RowType buildBinlogRowType( return new org.apache.flink.table.types.logical.RowType(fields); } + + /** + * Builds the produced Flink RowType for a binlog scan: the full binlog row type when {@code + * projectedTopLevel} is null, otherwise the projected top-level subset selected in the + * requested order. + * + * @param originalDataType the data columns row type + * @param projectedTopLevel top-level projection over the binlog row, or {@code null} for none + * @return the produced row type + */ + public static org.apache.flink.table.types.logical.RowType buildProducedRowType( + org.apache.flink.table.types.logical.RowType originalDataType, + @Nullable int[] projectedTopLevel) { + org.apache.flink.table.types.logical.RowType full = buildBinlogRowType(originalDataType); + if (projectedTopLevel == null) { + return full; + } + List fields = new ArrayList<>(); + for (int idx : projectedTopLevel) { + fields.add(full.getFields().get(idx)); + } + return new org.apache.flink.table.types.logical.RowType(fields); + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java index a2b339da5d2..508f4e36de8 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java @@ -27,11 +27,14 @@ import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.data.utils.JoinedRowData; +import org.apache.flink.table.data.utils.ProjectedRowData; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.types.RowKind; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; @@ -44,10 +47,30 @@ public class ChangelogRowConverter implements RecordToFlinkRowConverter { private final FlussRowToFlinkRowConverter baseConverter; private final org.apache.flink.table.types.logical.RowType producedType; - /** Creates a new ChangelogRowConverter. */ + /** + * Optional projection over the base changelog row {@code [_change_type, _log_offset, + * _commit_timestamp, ]}. When non-null, each entry is a position in the + * base row and the converter emits exactly those columns in that order; {@code null} means the + * full changelog row is emitted. + */ + @Nullable private final int[] baseRowProjection; + + /** Creates a new ChangelogRowConverter without projection. */ public ChangelogRowConverter(RowType rowType) { + this(rowType, null); + } + + /** + * Creates a new ChangelogRowConverter. + * + * @param rowType the (possibly projected) data columns scanned from Fluss + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + */ + public ChangelogRowConverter(RowType rowType, @Nullable int[] baseRowProjection) { this.baseConverter = new FlussRowToFlinkRowConverter(rowType); - this.producedType = buildChangelogRowType(FlinkConversions.toFlinkRowType(rowType)); + this.baseRowProjection = baseRowProjection; + this.producedType = + buildProducedRowType(FlinkConversions.toFlinkRowType(rowType), baseRowProjection); } /** Converts a LogRecord to a Flink RowData with metadata columns. */ @@ -71,7 +94,10 @@ public RowData toChangelogRowData(LogRecord record) { JoinedRowData joinedRow = new JoinedRowData(metadataRow, physicalRowData); joinedRow.setRowKind(RowKind.INSERT); - return joinedRow; + if (baseRowProjection == null) { + return joinedRow; + } + return ProjectedRowData.from(baseRowProjection).replaceRow(joinedRow); } @Override @@ -128,4 +154,27 @@ public static org.apache.flink.table.types.logical.RowType buildChangelogRowType return new org.apache.flink.table.types.logical.RowType(fields); } + + /** + * Builds the produced Flink RowType for a changelog scan: the full changelog row type (metadata + * columns prepended to {@code originalDataType}) when {@code baseRowProjection} is null, + * otherwise the projected subset selected in the requested order. + * + * @param originalDataType the (possibly projected) data columns row type + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + * @return the produced row type + */ + public static org.apache.flink.table.types.logical.RowType buildProducedRowType( + org.apache.flink.table.types.logical.RowType originalDataType, + @Nullable int[] baseRowProjection) { + org.apache.flink.table.types.logical.RowType full = buildChangelogRowType(originalDataType); + if (baseRowProjection == null) { + return full; + } + List fields = new ArrayList<>(); + for (int idx : baseRowProjection) { + fields.add(full.getFields().get(idx)); + } + return new org.apache.flink.table.types.logical.RowType(fields); + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java index 0c6be61d829..1dde6bf3da3 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java @@ -26,6 +26,8 @@ import org.apache.fluss.client.table.scanner.batch.BatchScanner; import org.apache.fluss.client.table.writer.UpsertWriter; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.StatisticsColumnsConfig; +import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.flink.source.lookup.FlinkLookupFunction; @@ -35,12 +37,17 @@ import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.metadata.TableStats; +import org.apache.fluss.predicate.CompoundPredicate; +import org.apache.fluss.predicate.LeafPredicate; +import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.predicate.PredicateVisitor; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.Decimal; import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.TimestampLtz; import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.types.DataTypeChecks; import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.DecimalData; @@ -57,6 +64,8 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import javax.annotation.Nullable; @@ -67,8 +76,10 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -79,6 +90,8 @@ /** Utilities for pushdown abilities. */ public class PushdownUtils { + private static final Logger LOG = LoggerFactory.getLogger(PushdownUtils.class); + private static final int MAX_LIMIT_PUSHDOWN = 2048; /** Extract field equality information from expressions. */ @@ -450,6 +463,86 @@ private static ListOffsetsResult listOffsets( // ------------------------------------------------------------------------------------------ + /** + * Computes the set of column names that have statistics available for filter pushdown, based on + * the table's statistics configuration. Returns an empty set when statistics collection is + * disabled. + * + * @param flussRowType the row type whose columns are considered + * @param tableConfig the table configuration carrying the statistics settings + * @return the set of column names that have statistics available + */ + public static Set computeAvailableStatsColumns( + org.apache.fluss.types.RowType flussRowType, TableConfig tableConfig) { + StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns(); + if (!statsConfig.isEnabled()) { + LOG.debug("Statistics collection is disabled for the table"); + return Collections.emptySet(); + } + + Set columns = new HashSet<>(); + if (statsConfig.getMode() == StatisticsColumnsConfig.Mode.ALL) { + for (int i = 0; i < flussRowType.getFieldCount(); i++) { + if (DataTypeChecks.isSupportedStatisticsType(flussRowType.getTypeAt(i))) { + columns.add(flussRowType.getFieldNames().get(i)); + } + } + } else { + for (String columnName : statsConfig.getColumns()) { + int columnIndex = flussRowType.getFieldNames().indexOf(columnName); + if (columnIndex >= 0) { + if (DataTypeChecks.isSupportedStatisticsType( + flussRowType.getTypeAt(columnIndex))) { + columns.add(columnName); + } else { + LOG.trace( + "Configured statistics column '{}' has unsupported type and will be ignored", + columnName); + } + } else { + LOG.trace( + "Configured statistics column '{}' does not exist in table schema", + columnName); + } + } + } + return columns; + } + + /** + * Checks if a predicate can benefit from statistics based on the available statistics columns. + * A compound predicate is usable only if all of its children are usable. + * + * @param predicate the predicate to check + * @param rowType the row type the predicate indexes into + * @param availableStatsColumns the columns that have statistics available + * @return true if the predicate can use statistics + */ + public static boolean canPredicateUseStatistics( + Predicate predicate, + org.apache.fluss.types.RowType rowType, + Set availableStatsColumns) { + class StatisticsUsageVisitor implements PredicateVisitor { + @Override + public Boolean visit(LeafPredicate leaf) { + String fieldName = rowType.getFieldNames().get(leaf.index()); + return availableStatsColumns.contains(fieldName); + } + + @Override + public Boolean visit(CompoundPredicate compound) { + for (Predicate child : compound.children()) { + if (!child.visit(this)) { + return false; + } + } + return true; + } + } + + return predicate.visit(new StatisticsUsageVisitor()); + } + /** A structure represents a source field equal literal expression. */ public static class FieldEqual implements Serializable { private static final long serialVersionUID = 1L; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java new file mode 100644 index 00000000000..b5f8f5e4ab3 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java @@ -0,0 +1,131 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.utils.BinlogRowConverter; +import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.FieldReferenceExpression; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link BinlogFlinkTableSource} projection and filter pushdown. */ +class BinlogFlinkTableSourceTest { + + // Data columns: (id INT, name STRING, amount BIGINT) + private static final RowType DATA_COLUMNS_TYPE = + (RowType) + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.INT()), + DataTypes.FIELD("name", DataTypes.STRING()), + DataTypes.FIELD("amount", DataTypes.BIGINT())) + .getLogicalType(); + + private BinlogFlinkTableSource createSource() { + FlinkConnectorOptionsUtils.StartupOptions startupOptions = + new FlinkConnectorOptionsUtils.StartupOptions(); + startupOptions.startupMode = ScanStartupMode.EARLIEST; + return new BinlogFlinkTableSource( + TablePath.of("db", "t"), + new Configuration(), + BinlogRowConverter.buildBinlogRowType(DATA_COLUMNS_TYPE), + false, + true, + startupOptions, + 1000L, + Collections.emptyMap()); + } + + private DataType projectedType(int... topLevelIndexes) { + RowType full = BinlogRowConverter.buildBinlogRowType(DATA_COLUMNS_TYPE); + DataTypes.Field[] fields = new DataTypes.Field[topLevelIndexes.length]; + for (int i = 0; i < topLevelIndexes.length; i++) { + RowType.RowField f = full.getFields().get(topLevelIndexes[i]); + fields[i] = + DataTypes.FIELD( + f.getName(), + org.apache.flink.table.types.utils.TypeConversions + .fromLogicalToDataType(f.getType())); + } + return DataTypes.ROW(fields); + } + + private int[][] nested(int... indexes) { + int[][] result = new int[indexes.length][]; + for (int i = 0; i < indexes.length; i++) { + result[i] = new int[] {indexes[i]}; + } + return result; + } + + @Test + void testApplyProjectionStoresTopLevel() { + BinlogFlinkTableSource source = createSource(); + // virtual [_change_type(0), after(4)] + source.applyProjection(nested(0, 4), projectedType(0, 4)); + + assertThat(source.getProjectedTopLevel()).containsExactly(0, 4); + assertThat(source.getProducedDataType()).isEqualTo(projectedType(0, 4).getLogicalType()); + } + + @Test + void testCopyPreservesPushdownState() { + BinlogFlinkTableSource source = createSource(); + source.applyProjection(nested(3, 4), projectedType(3, 4)); + + BinlogFlinkTableSource copy = (BinlogFlinkTableSource) source.copy(); + assertThat(copy.getProjectedTopLevel()).containsExactly(3, 4); + assertThat(copy.getProducedDataType()).isEqualTo(source.getProducedDataType()); + } + + @Test + void testApplyFiltersReturnsAllRemaining() { + BinlogFlinkTableSource source = createSource(); + FieldReferenceExpression ref = + new FieldReferenceExpression("_change_type", DataTypes.STRING(), 0, 0); + ValueLiteralExpression lit = + new ValueLiteralExpression("insert", DataTypes.STRING().notNull()); + ResolvedExpression filter = + CallExpression.permanent( + BuiltInFunctionDefinitions.EQUALS, + Arrays.asList(ref, lit), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + // Binlog data/partition columns are nested and metadata is non-pushable: nothing accepted. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java index 4787fe162da..53f50855bab 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java @@ -345,6 +345,74 @@ public void testBinlogSelectStar() throws Exception { .isEqualTo("+I[insert, 0, 1970-01-01T00:00:01Z, null, +I[1, Alice]]"); } + @Test + public void testBinlogTopLevelProjection() throws Exception { + // Select only the top-level columns _change_type and after (skip _log_offset, + // _commit_timestamp and before). The data scan stays full, so the nested after ROW still + // carries all original columns. + tEnv.executeSql( + "CREATE TABLE binlog_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "binlog_projection_test"); + + String query = "SELECT _change_type, after FROM binlog_projection_test$binlog"; + + // The top-level projection should be pushed into the source scan (assert the exact + // pushed projection rather than just the presence of "project=["). + assertThat(tEnv.explainSql(query)).contains("project=[_change_type, after]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(1000)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false); + + List results = collectRowsWithTimeout(rowIter, 1, true); + assertThat(results).hasSize(1); + // after ROW retains all data columns even though only two top-level columns are projected. + assertThat(results.get(0)).isEqualTo("+I[insert, +I[1, Alice, 100]]"); + rowIter.close(); + } + + @Test + public void testBinlogReorderedTopLevelProjection() throws Exception { + // Reordered top-level projection: after before _change_type, i.e. not in the declared + // order of the binlog row. This exercises the projectedTopLevel handling + // (applyProjection -> BinlogRowConverter/ProjectedRowData) end to end through the real + // planner, which no ascending-order projection test covers. + tEnv.executeSql( + "CREATE TABLE binlog_reordered_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "binlog_reordered_projection_test"); + + String query = "SELECT after, _change_type FROM binlog_reordered_projection_test$binlog"; + // The planner pushes the reordered projection into the scan as-is (no Calc reorder), so + // the source receives the out-of-order indices end to end. + assertThat(tEnv.explainSql(query)).contains("project=[after, _change_type]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false); + List insertResult = collectRowsWithTimeout(rowIter, 1, false); + assertThat(insertResult).containsExactly("+I[+I[1, Alice, 100], insert]"); + + // An update exercises the -U/+U merge together with the reordered projection. + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 250L)), false); + List updateResult = collectRowsWithTimeout(rowIter, 1, true); + assertThat(updateResult).containsExactly("+I[+I[1, Alice, 250], update]"); + } + @Test public void testBinlogWithPartitionedTable() throws Exception { // Create a partitioned primary key table diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java new file mode 100644 index 00000000000..a4528ef9d74 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java @@ -0,0 +1,306 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.utils.ChangelogRowConverter; +import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.FieldReferenceExpression; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; + +import static org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link ChangelogFlinkTableSource} projection and filter pushdown. */ +class ChangelogFlinkTableSourceTest { + + // Data columns: (id INT, name STRING, amount BIGINT) + private static final RowType DATA_COLUMNS_TYPE = + (RowType) + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.INT()), + DataTypes.FIELD("name", DataTypes.STRING()), + DataTypes.FIELD("amount", DataTypes.BIGINT())) + .getLogicalType(); + + private ChangelogFlinkTableSource createSource(int[] partitionKeyIndexes) { + return createSource(partitionKeyIndexes, Collections.emptyMap()); + } + + private ChangelogFlinkTableSource createSource( + int[] partitionKeyIndexes, Map extraOptions) { + FlinkConnectorOptionsUtils.StartupOptions startupOptions = + new FlinkConnectorOptionsUtils.StartupOptions(); + startupOptions.startupMode = ScanStartupMode.EARLIEST; + Map tableOptions = new HashMap<>(extraOptions); + return new ChangelogFlinkTableSource( + TablePath.of("db", "t"), + new Configuration(), + ChangelogRowConverter.buildChangelogRowType(DATA_COLUMNS_TYPE), + partitionKeyIndexes, + true, + startupOptions, + 1000L, + tableOptions); + } + + private DataType projectedType(int... virtualIndexes) { + RowType full = ChangelogRowConverter.buildChangelogRowType(DATA_COLUMNS_TYPE); + DataTypes.Field[] fields = new DataTypes.Field[virtualIndexes.length]; + for (int i = 0; i < virtualIndexes.length; i++) { + RowType.RowField f = full.getFields().get(virtualIndexes[i]); + fields[i] = + DataTypes.FIELD( + f.getName(), + org.apache.flink.table.types.utils.TypeConversions + .fromLogicalToDataType(f.getType())); + } + return DataTypes.ROW(fields); + } + + private int[][] nested(int... indexes) { + int[][] result = new int[indexes.length][]; + for (int i = 0; i < indexes.length; i++) { + result[i] = new int[] {indexes[i]}; + } + return result; + } + + private CallExpression equals(String field, DataType fieldType, Object value) { + return call(BuiltInFunctionDefinitions.EQUALS, field, fieldType, value); + } + + private CallExpression call( + org.apache.flink.table.functions.BuiltInFunctionDefinition func, + String field, + DataType fieldType, + Object value) { + FieldReferenceExpression ref = new FieldReferenceExpression(field, fieldType, 0, 0); + ValueLiteralExpression lit = new ValueLiteralExpression(value, fieldType.notNull()); + return CallExpression.permanent(func, Arrays.asList(ref, lit), DataTypes.BOOLEAN()); + } + + @ParameterizedTest + @MethodSource("projectionCases") + void testApplyProjection( + String description, + int[] virtual, + int[] expectedDataProjection, + int[] expectedBaseProjection) { + ChangelogFlinkTableSource source = createSource(new int[0]); + source.applyProjection(nested(virtual), projectedType(virtual)); + + assertThat(source.getProjectedFields()).containsExactly(virtual); + + // dataProjection null for metadata-only projection + if (expectedDataProjection == null) { + assertThat(source.getDataProjection()).isNull(); + } else { + assertThat(source.getDataProjection()).containsExactly(expectedDataProjection); + } + + assertThat(source.getBaseRowProjection()).containsExactly(expectedBaseProjection); + assertThat(source.getProducedDataType()).isEqualTo(projectedType(virtual).getLogicalType()); + } + + private static Stream projectionCases() { + return Stream.of( + // description, virtual, expectedDataProjection, expectedBaseProjection + // metadata only + Arguments.of("metadata-only", new int[] {0}, null, new int[] {0}), + // data only + Arguments.of("data-only", new int[] {3, 5}, new int[] {0, 2}, new int[] {3, 4}), + // reordered mix + Arguments.of( + "reorderedMix", + new int[] {0, 5, 3}, + new int[] {2, 0}, + new int[] {0, 3, 4})); + } + + @Test + void testCopyPreservesPushdownState() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + source.applyProjection(nested(0, 3), projectedType(0, 3)); + source.applyFilters(Collections.singletonList(equals("amount", DataTypes.BIGINT(), 100L))); + + ChangelogFlinkTableSource copy = (ChangelogFlinkTableSource) source.copy(); + assertThat(copy.getProjectedFields()).containsExactly(0, 3); + assertThat(copy.getDataProjection()).containsExactly(0); + assertThat(copy.getBaseRowProjection()).containsExactly(0, 3); + assertThat(copy.getProducedDataType()).isEqualTo(source.getProducedDataType()); + assertThat(copy.getLogRecordBatchFilter()).isEqualTo(source.getLogRecordBatchFilter()); + } + + @Test + void testApplyFiltersDataColumnPushed() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression filter = equals("amount", DataTypes.BIGINT(), 100L); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).containsExactly(filter); + // All filters are returned as remaining (safety net). + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNotNull(); + } + + @Test + void testApplyFiltersDataColumnNotInStatisticsNotPushed() { + // Statistics are collected only for "amount". A filter on "name" cannot use batch + // statistics, so it must not be pushed as a record-batch filter even though it is a + // convertible data-column predicate. + ChangelogFlinkTableSource source = createSource(new int[0], statsColumns("amount")); + ResolvedExpression filter = equals("name", DataTypes.STRING(), "p1"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersCompoundDataColumnsPushed() { + // WHERE amount > 100 AND amount < 500: both leaves reference a statistics-backed data + // column, so the whole AND predicate is pushed as a single record-batch filter. + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression compound = + CallExpression.permanent( + BuiltInFunctionDefinitions.AND, + Arrays.asList( + call( + BuiltInFunctionDefinitions.GREATER_THAN, + "amount", + DataTypes.BIGINT(), + 100L), + call( + BuiltInFunctionDefinitions.LESS_THAN, + "amount", + DataTypes.BIGINT(), + 500L)), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(compound)); + + assertThat(result.getAcceptedFilters()).containsExactly(compound); + assertThat(result.getRemainingFilters()).containsExactly(compound); + assertThat(source.getLogRecordBatchFilter()).isNotNull(); + } + + @Test + void testApplyFiltersMetadataColumnNotPushed() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression filter = equals("_change_type", DataTypes.STRING(), "insert"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersPartitionKeyPushed() { + // partition key = data column "name" (index 1) + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression filter = equals("name", DataTypes.STRING(), "p1"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).containsExactly(filter); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getPartitionFilters()).isNotNull(); + } + + @Test + void testApplyFiltersNonPartitionKeyNotPushedToPartitionFilters() { + // partition key = data column "name" (index 1); the filter references the non-partition + // column "amount", so it must not create a partition filter and must stay a post-filter. + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression filter = equals("amount", DataTypes.BIGINT(), 100L); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + // Statistics are disabled by default, so the filter is not accepted as a batch filter + // either. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getPartitionFilters()).isNull(); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersCompoundPartitionAndNonPartitionNotPushed() { + // partition key = data column "name" (index 1). A compound filter referencing both a + // partition column and a non-partition column cannot be used for partition pruning as a + // whole; it must remain a post-filter so no condition is silently dropped. + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression compound = + CallExpression.permanent( + BuiltInFunctionDefinitions.AND, + Arrays.asList( + equals("name", DataTypes.STRING(), "p1"), + equals("amount", DataTypes.BIGINT(), 100L)), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(compound)); + + // Statistics are disabled by default, so nothing is accepted at all. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(compound); + assertThat(source.getPartitionFilters()).isNull(); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + private Map statsAll() { + return statsColumns("*"); + } + + private Map statsColumns(String columns) { + Map options = new HashMap<>(); + options.put("table.statistics.columns", columns); + return options; + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index d9fbf9e5c1d..ea5446604c6 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -344,6 +344,142 @@ public void testProjectionOnChangelogTable() throws Exception { assertThat(deleteResult.get(0)).isEqualTo("+I[delete, 1, Item-1-Updated]"); } + @Test + public void testReorderedProjectionOnChangelogTable() throws Exception { + // Reordered projection: data columns out of their original order with a metadata column + // in between. This exercises the dataProjection/baseRowProjection coordinate translation + // (applyProjection -> ChangelogRowConverter) end to end through the real planner. + tEnv.executeSql( + "CREATE TABLE reordered_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "reordered_projection_test"); + + String query = "SELECT amount, _change_type, id FROM reordered_projection_test$changelog"; + // The planner pushes the reordered projection into the scan as-is (no Calc reorder), so + // the source receives the out-of-order indices end to end. + assertThat(tEnv.explainSql(query)).contains("project=[amount, _change_type, id]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1", 100L)), false); + List insertResult = collectRowsWithTimeout(rowIter, 1, false); + assertThat(insertResult).containsExactly("+I[100, insert, 1]"); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1", 250L)), false); + List updateResults = collectRowsWithTimeout(rowIter, 2, true); + assertThat(updateResults) + .containsExactly("+I[100, update_before, 1]", "+I[250, update_after, 1]"); + } + + @Test + public void testChangelogDataColumnFilterAndProjectionPushdown() throws Exception { + // Statistics columns are only supported on log tables (not PK tables), so use a log table. + // A log table's changelog is append-only, which is sufficient to exercise pushdown. + tEnv.executeSql( + "CREATE TABLE data_filter_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT" + + ") WITH ('bucket.num' = '1', 'table.statistics.columns' = 'amount')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "data_filter_test"); + + String query = + "SELECT _change_type, id, amount FROM data_filter_test$changelog WHERE amount > 150"; + + // Projection and the data-column filter should be pushed into the source scan. Assert the + // exact pushed projection and predicate, because "filter=[" would also match an empty + // "filter=[]" digest when nothing is pushed down. + String plan = tEnv.explainSql(query); + assertThat(plan).contains("project=[_change_type, id, amount]"); + assertThat(plan).contains("filter=[>(amount, 150)]"); + // The filter is still retained in the Calc operator as a safety net (FLINK-38635). + assertThat(plan).contains("where="); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows( + conn, + tablePath, + Arrays.asList( + row(1, "Item-1", 100L), row(2, "Item-2", 200L), row(3, "Item-3", 300L)), + true); + + // Only amount > 150 rows pass the filter (safety net + pushdown produce identical results). + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results).containsExactly("+I[insert, 2, 200]", "+I[insert, 3, 300]"); + rowIter.close(); + } + + @Test + public void testChangelogPartitionFilterPushdown() throws Exception { + tEnv.executeSql( + "CREATE TABLE partition_filter_test (" + + " id INT NOT NULL," + + " name STRING," + + " region STRING NOT NULL," + + " PRIMARY KEY (id, region) NOT ENFORCED" + + ") PARTITIONED BY (region) WITH ('bucket.num' = '1')"); + + CLOCK.advanceTime(Duration.ofMillis(100)); + tEnv.executeSql( + "INSERT INTO partition_filter_test VALUES " + + "(1, 'Item-1', 'us'), " + + "(2, 'Item-2', 'us'), " + + "(3, 'Item-3', 'eu')") + .await(); + + String query = + "SELECT _change_type, id, name, region FROM partition_filter_test$changelog " + + "WHERE region = 'us'"; + + // The partition-key filter should be pushed into the source scan (assert the exact pushed + // predicate), and retained in the Calc operator as a safety net (FLINK-38635). + String plan = tEnv.explainSql(query); + assertThat(plan).contains("filter=[=(region, _UTF-16LE'us'"); + assertThat(plan).contains("where="); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results) + .containsExactly("+I[insert, 1, Item-1, us]", "+I[insert, 2, Item-2, us]"); + rowIter.close(); + } + + @Test + public void testChangelogMetadataOnlyProjection() throws Exception { + // Projecting only metadata columns leaves no data column to scan. The source normalizes the + // scan projection to null (full data row scan) to avoid a zero-column scan that produces no + // records, then emits only the metadata columns. Verify this runtime path works end to end. + tEnv.executeSql( + "CREATE TABLE metadata_only_test (" + + " id INT NOT NULL," + + " name STRING" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "metadata_only_test"); + + String query = "SELECT _change_type, _log_offset FROM metadata_only_test$changelog"; + assertThat(tEnv.explainSql(query)).contains("project=[_change_type, _log_offset]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1"), row(2, "Item-2")), true); + + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results).containsExactly("+I[insert, 0]", "+I[insert, 1]"); + rowIter.close(); + } + @Test public void testChangelogScanWithAllChangeTypes() throws Exception { // Create a primary key table with 1 bucket for consistent log_offset numbers @@ -539,10 +675,11 @@ public void testChangelogWithPartitionedTable() throws Exception { String query = "SELECT _change_type, id, name, region FROM partitioned_test$changelog"; CloseableIterator rowIter = tEnv.executeSql(query).collect(); - // Collect initial inserts + // Collect initial inserts. Records from different partitions may arrive in any order, + // so the assertion must be order-insensitive. List results = collectRowsWithTimeout(rowIter, 3, false); assertThat(results) - .containsExactly( + .containsExactlyInAnyOrder( "+I[insert, 1, Item-1, us]", "+I[insert, 2, Item-2, us]", "+I[insert, 3, Item-3, eu]"); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java index a09dd7c4199..ee0ab4e67e7 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java @@ -229,6 +229,71 @@ void testMultipleUpdatesInSequence() throws Exception { assertThat(result2.getLong(1)).isEqualTo(20L); // offset from second -U } + @Test + void testProjectionEmitsSelectedTopLevelColumns() throws Exception { + // Project the binlog row [_change_type(0), _log_offset(1), _commit_timestamp(2), + // before(3), after(4)] down to [_change_type, after]. The data scan stays full, so the + // nested after ROW still contains all columns. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {0, 4}); + + org.apache.flink.table.types.logical.RowType producedType = projected.getProducedType(); + assertThat(producedType.getFieldNames()).containsExactly("_change_type", "after"); + assertThat(producedType.getTypeAt(1)) + .isInstanceOf(org.apache.flink.table.types.logical.RowType.class); + + RowData result = + projected.convert( + createLogRecord(ChangeType.INSERT, 100L, 1000L, 1, "Alice", 5000L)); + + assertThat(result.getArity()).isEqualTo(2); + assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); + assertThat(result.getString(0)).isEqualTo(StringData.fromString("insert")); + RowData afterRow = result.getRow(1, 3); + assertThat(afterRow.getInt(0)).isEqualTo(1); + assertThat(afterRow.getString(1).toString()).isEqualTo("Alice"); + assertThat(afterRow.getLong(2)).isEqualTo(5000L); + } + + @Test + void testProjectionUpdateMerge() throws Exception { + // Projection must not disturb the -U/+U buffering: project [before, after]. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {3, 4}); + + assertThat( + projected.convert( + createLogRecord( + ChangeType.UPDATE_BEFORE, 200L, 2000L, 2, "Bob", 3000L))) + .isNull(); + RowData result = + projected.convert( + createLogRecord( + ChangeType.UPDATE_AFTER, 201L, 2000L, 2, "Bob-Updated", 4000L)); + + assertThat(result.getArity()).isEqualTo(2); + RowData beforeRow = result.getRow(0, 3); + assertThat(beforeRow.getString(1).toString()).isEqualTo("Bob"); + RowData afterRow = result.getRow(1, 3); + assertThat(afterRow.getString(1).toString()).isEqualTo("Bob-Updated"); + } + + @Test + void testProjectionDeleteRecord() throws Exception { + // Project [_change_type, before, after]. For a DELETE the after image is null and must + // pass through the projection untouched. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {0, 3, 4}); + + RowData result = + projected.convert( + createLogRecord(ChangeType.DELETE, 300L, 3000L, 3, "Charlie", 1000L)); + + assertThat(result.getArity()).isEqualTo(3); + assertThat(result.getString(0)).isEqualTo(StringData.fromString("delete")); + RowData beforeRow = result.getRow(1, 3); + assertThat(beforeRow.getInt(0)).isEqualTo(3); + assertThat(beforeRow.getString(1).toString()).isEqualTo("Charlie"); + assertThat(result.isNullAt(2)).isTrue(); + } + private LogRecord createLogRecord( ChangeType changeType, long offset, long timestamp, int id, String name, long amount) throws Exception { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java index 55f83a51498..5a31fbd591d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java @@ -33,6 +33,11 @@ import org.apache.flink.types.RowKind; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -55,17 +60,19 @@ void setUp() { converter = new ChangelogRowConverter(testRowType); } - @Test - void testConvertInsertRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.INSERT, 100L, 1, "Alice", 5000L); + @ParameterizedTest + @MethodSource("changeTypeCases") + void testConvertAllChangeTypes(ChangeType changeType, String expectedChangeType) + throws Exception { + LogRecord record = createLogRecord(changeType, 100L, 1, "Alice", 5000L); RowData result = converter.convert(record); - // Verify row kind + // Virtual table always emits INSERT row kind, regardless of the underlying change type. assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); // Verify metadata columns - assertThat(result.getString(0)).isEqualTo(StringData.fromString("insert")); + assertThat(result.getString(0)).isEqualTo(StringData.fromString(expectedChangeType)); assertThat(result.getLong(1)).isEqualTo(100L); // log offset assertThat(result.getTimestamp(2, 3)).isNotNull(); // commit timestamp @@ -78,111 +85,47 @@ void testConvertInsertRecord() throws Exception { assertThat(result).isInstanceOf(JoinedRowData.class); } - @Test - void testConvertUpdateBeforeRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.UPDATE_BEFORE, 200L, 2, "Bob", 3000L); - - RowData result = converter.convert(record); - - // Verify row kind (always INSERT for virtual table) - assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); - - // Verify change type metadata - assertThat(result.getString(0)).isEqualTo(StringData.fromString("update_before")); - assertThat(result.getLong(1)).isEqualTo(200L); - - // Verify physical columns - assertThat(result.getInt(3)).isEqualTo(2); - assertThat(result.getString(4).toString()).isEqualTo("Bob"); - assertThat(result.getLong(5)).isEqualTo(3000L); + private static Stream changeTypeCases() { + return Stream.of( + Arguments.of(ChangeType.INSERT, "insert"), + Arguments.of(ChangeType.DELETE, "delete"), + Arguments.of(ChangeType.UPDATE_BEFORE, "update_before"), + Arguments.of(ChangeType.UPDATE_AFTER, "update_after"), + Arguments.of(ChangeType.APPEND_ONLY, "insert")); } @Test - void testConvertUpdateAfterRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.UPDATE_AFTER, 201L, 2, "Bob", 4000L); + void testProjectionEmitsSelectedColumnsInOrder() throws Exception { + // Project the base changelog row + // [_change_type(0), _log_offset(1), _commit_timestamp(2), id(3), name(4), amount(5)] + // down to [_log_offset, amount, id] to verify subsetting + reordering. + ChangelogRowConverter projected = + new ChangelogRowConverter(testRowType, new int[] {1, 5, 3}); - RowData result = converter.convert(record); + org.apache.flink.table.types.logical.RowType producedType = projected.getProducedType(); + assertThat(producedType.getFieldNames()).containsExactly("_log_offset", "amount", "id"); + + RowData result = + projected.convert(createLogRecord(ChangeType.INSERT, 100L, 1, "Alice", 5000L)); - assertThat(result.getString(0)).isEqualTo(StringData.fromString("update_after")); - assertThat(result.getLong(1)).isEqualTo(201L); - assertThat(result.getInt(3)).isEqualTo(2); - assertThat(result.getString(4).toString()).isEqualTo("Bob"); - assertThat(result.getLong(5)).isEqualTo(4000L); + assertThat(result.getArity()).isEqualTo(3); + assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); + assertThat(result.getLong(0)).isEqualTo(100L); // _log_offset + assertThat(result.getLong(1)).isEqualTo(5000L); // amount + assertThat(result.getInt(2)).isEqualTo(1); // id } @Test - void testConvertDeleteRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.DELETE, 300L, 3, "Charlie", 1000L); + void testProjectionMetadataOnly() throws Exception { + // Project only the _change_type metadata column. + ChangelogRowConverter projected = new ChangelogRowConverter(testRowType, new int[] {0}); - RowData result = converter.convert(record); + assertThat(projected.getProducedType().getFieldNames()).containsExactly("_change_type"); + RowData result = + projected.convert(createLogRecord(ChangeType.DELETE, 300L, 3, "Charlie", 1000L)); + assertThat(result.getArity()).isEqualTo(1); assertThat(result.getString(0)).isEqualTo(StringData.fromString("delete")); - assertThat(result.getLong(1)).isEqualTo(300L); - assertThat(result.getInt(3)).isEqualTo(3); - assertThat(result.getString(4).toString()).isEqualTo("Charlie"); - assertThat(result.getLong(5)).isEqualTo(1000L); - } - - @Test - void testProducedTypeHasMetadataColumns() { - org.apache.flink.table.types.logical.RowType producedType = converter.getProducedType(); - - // Should have 3 metadata columns + 3 physical columns - assertThat(producedType.getFieldCount()).isEqualTo(6); - - // Check metadata column names and types - assertThat(producedType.getFieldNames()) - .containsExactly( - "_change_type", "_log_offset", "_commit_timestamp", "id", "name", "amount"); - - // Check metadata column types - assertThat(producedType.getTypeAt(0)) - .isInstanceOf(org.apache.flink.table.types.logical.VarCharType.class); - assertThat(producedType.getTypeAt(1)) - .isInstanceOf(org.apache.flink.table.types.logical.BigIntType.class); - assertThat(producedType.getTypeAt(2)) - .isInstanceOf(org.apache.flink.table.types.logical.LocalZonedTimestampType.class); - } - - @Test - void testAllChangeTypes() throws Exception { - // Test all change type conversions - assertThat( - converter - .convert(createLogRecord(ChangeType.INSERT, 1L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("insert")); - - assertThat( - converter - .convert( - createLogRecord( - ChangeType.UPDATE_BEFORE, 2L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("update_before")); - - assertThat( - converter - .convert( - createLogRecord( - ChangeType.UPDATE_AFTER, 3L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("update_after")); - - assertThat( - converter - .convert(createLogRecord(ChangeType.DELETE, 4L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("delete")); - - // For log tables (append-only) - assertThat( - converter - .convert( - createLogRecord( - ChangeType.APPEND_ONLY, 5L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("insert")); } private LogRecord createLogRecord(