diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotAndLogBatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotAndLogBatchScanner.java new file mode 100644 index 0000000000..88e72e42a0 --- /dev/null +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotAndLogBatchScanner.java @@ -0,0 +1,325 @@ +/* + * 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.client.table.scanner.batch; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.client.table.scanner.SortMergeReader; +import org.apache.fluss.client.table.scanner.log.LogScanner; +import org.apache.fluss.client.table.scanner.log.ScanRecords; +import org.apache.fluss.memory.MemorySegment; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.record.LogRecord; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.KeyValueRow; +import org.apache.fluss.row.encode.KeyEncoder; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.CloseableIterator; +import org.apache.fluss.utils.IOUtils; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * A bounded scanner that merges an immutable KV snapshot with a fixed primary-key changelog range. + */ +@Internal +public class KvSnapshotAndLogBatchScanner implements BatchScanner { + + private final TableBucket tableBucket; + private final long logStoppingOffset; + private final int[] keyIndexesInScanRow; + @Nullable private final int[] adjustProjectedFields; + private final Comparator primaryKeyComparator; + private final Map logRows; + + @Nullable private final BatchScanner snapshotScanner; + @Nullable private final LogScanner logScanner; + + private boolean logScanFinished; + private boolean snapshotIteratorInitialized; + private boolean finished; + private boolean closed; + + @Nullable private CloseableIterator snapshotRecordIterator; + @Nullable private SortMergeReader sortMergeReader; + @Nullable private CloseableIterator sortMergeIterator; + + public KvSnapshotAndLogBatchScanner( + Table table, + TableBucket tableBucket, + long snapshotId, + long logStartingOffset, + long logStoppingOffset, + @Nullable int[] projectedFields) { + checkArgument( + table.getTableInfo().hasPrimaryKey(), + "KvSnapshotAndLogBatchScanner only supports primary-key tables."); + this.tableBucket = tableBucket; + this.logStoppingOffset = logStoppingOffset; + + ProjectionPlan projectionPlan = createProjectionPlan(table.getTableInfo(), projectedFields); + this.keyIndexesInScanRow = projectionPlan.keyIndexesInScanRow; + this.adjustProjectedFields = projectionPlan.adjustProjectedFields; + this.primaryKeyComparator = createPrimaryKeyComparator(table.getTableInfo()); + this.logRows = new TreeMap<>(primaryKeyComparator); + + this.snapshotScanner = + snapshotId >= 0 + ? table.newScan() + .project(projectionPlan.scanProjectedFields) + .createBatchScanner(tableBucket, snapshotId) + : null; + + boolean emptyLogRange = logStartingOffset >= logStoppingOffset || logStoppingOffset <= 0; + this.logScanFinished = emptyLogRange; + if (emptyLogRange) { + this.logScanner = null; + } else { + this.logScanner = + table.newScan().project(projectionPlan.scanProjectedFields).createLogScanner(); + Long partitionId = tableBucket.getPartitionId(); + if (partitionId == null) { + this.logScanner.subscribe(tableBucket.getBucket(), logStartingOffset); + } else { + this.logScanner.subscribe(partitionId, tableBucket.getBucket(), logStartingOffset); + } + } + } + + @Nullable + @Override + public CloseableIterator pollBatch(Duration timeout) throws IOException { + if (closed || finished) { + return null; + } + + if (!logScanFinished) { + pollLogRecords(timeout); + return CloseableIterator.emptyIterator(); + } + + if (sortMergeReader == null) { + if (!initializeSnapshotIterator(timeout)) { + return CloseableIterator.emptyIterator(); + } + + sortMergeReader = + new SortMergeReader( + adjustProjectedFields, + keyIndexesInScanRow, + snapshotRecordIterator, + primaryKeyComparator, + CloseableIterator.wrap(logRows.values().iterator())); + } + + sortMergeIterator = sortMergeReader.readBatch(); + if (sortMergeIterator == null) { + finished = true; + } + return sortMergeIterator; + } + + private void pollLogRecords(Duration timeout) { + ScanRecords scanRecords = logScanner.poll(timeout); + for (ScanRecord scanRecord : scanRecords.records(tableBucket)) { + long logOffset = scanRecord.logOffset(); + if (logOffset >= logStoppingOffset) { + logScanFinished = true; + break; + } + + reduceLogRecord(scanRecord); + if (logOffset >= logStoppingOffset - 1) { + logScanFinished = true; + break; + } + } + + Long consumedUpToOffset = scanRecords.consumedUpToOffset(tableBucket); + if (consumedUpToOffset != null && consumedUpToOffset >= logStoppingOffset) { + logScanFinished = true; + } + } + + private void reduceLogRecord(ScanRecord scanRecord) { + ChangeType changeType = scanRecord.getChangeType(); + boolean isDelete = + changeType == ChangeType.DELETE || changeType == ChangeType.UPDATE_BEFORE; + KeyValueRow keyValueRow = + new KeyValueRow(keyIndexesInScanRow, scanRecord.getRow(), isDelete); + logRows.put(keyValueRow.keyRow(), keyValueRow); + } + + private boolean initializeSnapshotIterator(Duration timeout) throws IOException { + if (snapshotIteratorInitialized) { + return true; + } + + if (snapshotScanner == null) { + snapshotRecordIterator = CloseableIterator.emptyIterator(); + snapshotIteratorInitialized = true; + return true; + } + + CloseableIterator snapshotRows = snapshotScanner.pollBatch(timeout); + if (snapshotRows == null) { + snapshotRecordIterator = CloseableIterator.emptyIterator(); + snapshotIteratorInitialized = true; + return true; + } + + if (!snapshotRows.hasNext()) { + snapshotRows.close(); + return false; + } + + snapshotRecordIterator = new SnapshotRecordIterator(snapshotRows); + snapshotIteratorInitialized = true; + return true; + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + closed = true; + IOUtils.closeQuietly(sortMergeIterator); + IOUtils.closeQuietly(snapshotRecordIterator); + IOUtils.closeQuietly(snapshotScanner); + IOUtils.closeQuietly(logScanner); + } + + private static ProjectionPlan createProjectionPlan( + TableInfo tableInfo, @Nullable int[] projectedFields) { + int[] physicalPrimaryKeyIndexes = getPhysicalPrimaryKeyIndexes(tableInfo); + int fieldCount = tableInfo.getRowType().getFieldCount(); + if (projectedFields == null) { + return new ProjectionPlan( + IntStream.range(0, fieldCount).toArray(), physicalPrimaryKeyIndexes, null); + } + + List scanProjectedFields = + Arrays.stream(projectedFields).boxed().collect(Collectors.toList()); + int[] keyIndexesInScanRow = new int[physicalPrimaryKeyIndexes.length]; + for (int i = 0; i < physicalPrimaryKeyIndexes.length; i++) { + int primaryKeyIndex = physicalPrimaryKeyIndexes[i]; + int indexInProjectedFields = findIndex(projectedFields, primaryKeyIndex); + if (indexInProjectedFields >= 0) { + keyIndexesInScanRow[i] = indexInProjectedFields; + } else { + scanProjectedFields.add(primaryKeyIndex); + keyIndexesInScanRow[i] = scanProjectedFields.size() - 1; + } + } + + int[] scanProjection = scanProjectedFields.stream().mapToInt(Integer::intValue).toArray(); + int[] adjustProjectedFields = new int[projectedFields.length]; + for (int i = 0; i < projectedFields.length; i++) { + adjustProjectedFields[i] = findIndex(scanProjection, projectedFields[i]); + } + return new ProjectionPlan(scanProjection, keyIndexesInScanRow, adjustProjectedFields); + } + + private static Comparator createPrimaryKeyComparator(TableInfo tableInfo) { + int[] physicalPrimaryKeyIndexes = getPhysicalPrimaryKeyIndexes(tableInfo); + RowType primaryKeyRowType = + Schema.getKeyRowType(tableInfo.getSchema(), physicalPrimaryKeyIndexes); + KeyEncoder primaryKeyEncoder = + KeyEncoder.ofPrimaryKeyEncoder( + primaryKeyRowType, + tableInfo.getPhysicalPrimaryKeys(), + tableInfo.getTableConfig(), + tableInfo.isDefaultBucketKey()); + return (row1, row2) -> { + byte[] key1 = primaryKeyEncoder.encodeKey(row1); + byte[] key2 = primaryKeyEncoder.encodeKey(row2); + return MemorySegment.wrap(key1) + .compare(MemorySegment.wrap(key2), 0, 0, key1.length, key2.length); + }; + } + + private static int[] getPhysicalPrimaryKeyIndexes(TableInfo tableInfo) { + return tableInfo.getPhysicalPrimaryKeys().stream() + .mapToInt(primaryKey -> tableInfo.getRowType().getFieldIndex(primaryKey)) + .toArray(); + } + + private static int findIndex(int[] array, int target) { + for (int i = 0; i < array.length; i++) { + if (array[i] == target) { + return i; + } + } + return -1; + } + + private static class ProjectionPlan { + private final int[] scanProjectedFields; + private final int[] keyIndexesInScanRow; + @Nullable private final int[] adjustProjectedFields; + + private ProjectionPlan( + int[] scanProjectedFields, + int[] keyIndexesInScanRow, + @Nullable int[] adjustProjectedFields) { + this.scanProjectedFields = scanProjectedFields; + this.keyIndexesInScanRow = keyIndexesInScanRow; + this.adjustProjectedFields = adjustProjectedFields; + } + } + + private static class SnapshotRecordIterator implements CloseableIterator { + private final CloseableIterator rows; + + private SnapshotRecordIterator(CloseableIterator rows) { + this.rows = rows; + } + + @Override + public void close() { + rows.close(); + } + + @Override + public boolean hasNext() { + return rows.hasNext(); + } + + @Override + public LogRecord next() { + return new ScanRecord(rows.next()); + } + } +} 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 24f3d94d67..782c8c49f2 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 @@ -437,9 +437,12 @@ public boolean isBounded() { + modificationScanType + " statement with conditions on primary key."); } - if (!isDataLakeEnabled) { + if (hasPrimaryKey() + && startupOptions.startupMode + != FlinkConnectorOptions.ScanStartupMode.FULL) { throw new UnsupportedOperationException( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + "Currently, Fluss batch scan on primary-key tables only supports " + + "full startup mode."); } return source; } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java index 3280522738..392d77708f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java @@ -19,6 +19,7 @@ import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.client.initializer.OffsetsInitializer; +import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.FlinkConnectorOptions; import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema; @@ -114,7 +115,7 @@ public class FlussSource extends FlinkSource { sourceOutputType, projectedFields, logRecordBatchFilter, - offsetsInitializer, + validateBatchStartupMode(offsetsInitializer, hasPrimaryKey, streaming, tablePath), scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, deserializationSchema, @@ -133,6 +134,24 @@ public static FlussSourceBuilder builder() { return new FlussSourceBuilder<>(); } + private static OffsetsInitializer validateBatchStartupMode( + OffsetsInitializer offsetsInitializer, + boolean hasPrimaryKey, + boolean streaming, + TablePath tablePath) { + if (!streaming + && hasPrimaryKey + && !(offsetsInitializer instanceof SnapshotOffsetsInitializer)) { + throw new IllegalArgumentException( + String.format( + "Bounded (batch) read on primary-key tables requires full mode " + + "(OffsetsInitializer.full()), but table '%s' isn't started " + + "in full mode.", + tablePath)); + } + return offsetsInitializer; + } + @VisibleForTesting OffsetsInitializer getOffsetsInitializer() { return offsetsInitializer; diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java index 2e80067fed..f9ef727597 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java @@ -174,9 +174,11 @@ public FlussSourceBuilder setStartingOffsets(OffsetsInitializer offsetsInit /** * Builds a bounded source for batch execution. The source reads up to the latest offsets at job - * startup and then finishes; combined with the default {@link OffsetsInitializer#full()} on a - * datalake-enabled table this performs a bounded union read of the lake snapshot and the Fluss - * log. If not called, the source is unbounded (streaming). + * startup and then finishes. On datalake-enabled log tables, the default {@link + * OffsetsInitializer#full()} performs a bounded union read of the lake snapshot and the Fluss + * log. Primary-key tables require full startup mode for batch reads and use Fluss KV snapshots + * plus bounded logs. Log tables support all startup modes. If not called, the source is + * unbounded (streaming). * * @return this builder */ @@ -353,17 +355,19 @@ public FlussSource build() { boolean lakeEnabled = tableInfo.getTableConfig().isDataLakeEnabled(); boolean fullStartup = offsetsInitializer instanceof SnapshotOffsetsInitializer; - if (bounded && !(lakeEnabled && fullStartup)) { + if (bounded && hasPrimaryKey && !fullStartup) { throw new IllegalArgumentException( String.format( - "Bounded (batch) read requires a datalake-enabled table started in " - + "full mode (OffsetsInitializer.full()), but table '%s' has " - + "datalake enabled=%s and full startup mode=%s.", - tablePath, lakeEnabled, fullStartup)); + "Bounded (batch) read on primary-key tables requires full mode " + + "(OffsetsInitializer.full()), but table '%s' has full " + + "startup mode=%s.", + tablePath, fullStartup)); } LakeSource lakeSource = null; - if (lakeEnabled && fullStartup) { + // Bounded primary-key scans use Fluss KV snapshots plus bounded logs. Lake union-read is + // kept for log tables and streaming reads. + if (lakeEnabled && fullStartup && !(bounded && hasPrimaryKey)) { lakeSource = LakeSourceUtils.createLakeSource(tablePath, tableInfo.getProperties().toMap()); if (lakeSource != null) { diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java index 887eabf3b1..f774986ef2 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java @@ -26,7 +26,6 @@ import org.apache.fluss.client.initializer.OffsetsInitializer.BucketOffsetsRetriever; import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer; import org.apache.fluss.client.metadata.KvSnapshots; -import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.flink.FlinkConnectorOptions; @@ -575,11 +574,15 @@ public void start() { } private void startInBatchMode() { - if (lakeEnabled) { - if (lakeSource == null) { - throw new IllegalStateException( - "The 'lakeSource' is null in batch mode. It should be set if lake is enabled."); - } + if (hasPrimaryKey && !(startingOffsetsInitializer instanceof SnapshotOffsetsInitializer)) { + throw new UnsupportedOperationException( + "Batch mode on primary-key tables only supports full startup mode."); + } + + FlussOnlyBatchSplitGenerator flussOnlyBatchSplitGenerator = + createFlussOnlyBatchSplitGenerator(); + boolean useLakeUnionRead = lakeEnabled && lakeSource != null; + if (useLakeUnionRead) { context.callAsync( () -> { List splits = generateHybridLakeFlussSplits(); @@ -589,37 +592,29 @@ private void startInBatchMode() { "No lake snapshot found for table {}," + " falling back to Fluss-only splits.", tablePath); - if (isPartitioned) { - Set partitionInfos = listPartitions(); - Collection partitions = - partitionInfos.stream() - .map( - p -> - new Partition( - p.getPartitionId(), - p.getPartitionName())) - .collect(Collectors.toList()); - // Use log-only splits to avoid generating mixed split - // types (HybridSnapshotLogSplit + LogSplit) for - // primary-key tables, which is not supported. - splits = - this.initLogTablePartitionSplits( - partitions, startingOffsetsInitializer); - } else { - splits = this.getLogSplit(null, null); - } + splits = flussOnlyBatchSplitGenerator.generate(); } return splits; }, this::handleSplitsAdd); } else { - throw new UnsupportedOperationException( - String.format( - "Batch only supports when table option '%s' is set to true.", - ConfigOptions.TABLE_DATALAKE_ENABLED)); + context.callAsync(flussOnlyBatchSplitGenerator::generate, this::handleSplitsAdd); } } + private FlussOnlyBatchSplitGenerator createFlussOnlyBatchSplitGenerator() { + return new FlussOnlyBatchSplitGenerator( + tableInfo, + hasPrimaryKey, + isPartitioned, + startingOffsetsInitializer, + stoppingOffsetsInitializer, + bucketOffsetsRetriever, + this::listPartitions, + this::getLatestKvSnapshotsAndRegister, + this::ignoreTableBucket); + } + private void startInStreamModeForNonPartitionedTable() { // If we have restored unassigned splits from checkpoint state, skip re-initialization. // These splits already have their offsets resolved and will be assigned to readers @@ -990,7 +985,14 @@ private List getSnapshotAndLogSplits( "Log offset should be present if snapshot id is present."); splits.add( new HybridSnapshotLogSplit( - tb, partitionName, snapshotId.getAsLong(), logOffset.getAsLong())); + tb, + partitionName, + snapshotId.getAsLong(), + 0, + false, + logOffset.getAsLong(), + LogSplit.NO_STOPPING_OFFSET, + false)); } else { bucketsNeedInitOffset.add(bucketId); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlussOnlyBatchSplitGenerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlussOnlyBatchSplitGenerator.java new file mode 100644 index 0000000000..e17a57cdc1 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlussOnlyBatchSplitGenerator.java @@ -0,0 +1,214 @@ +/* + * 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.enumerator; + +import org.apache.fluss.client.initializer.OffsetsInitializer; +import org.apache.fluss.client.initializer.OffsetsInitializer.BucketOffsetsRetriever; +import org.apache.fluss.client.metadata.KvSnapshots; +import org.apache.fluss.client.table.scanner.log.LogScanner; +import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit; +import org.apache.fluss.flink.source.split.LogSplit; +import org.apache.fluss.flink.source.split.SourceSplitBase; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.OptionalLong; +import java.util.Set; +import java.util.function.Predicate; +import java.util.function.Supplier; + +import static org.apache.fluss.utils.Preconditions.checkState; + +/** Generates bounded Fluss-only splits for batch scans. */ +final class FlussOnlyBatchSplitGenerator { + + private final TableInfo tableInfo; + private final boolean hasPrimaryKey; + private final boolean isPartitioned; + private final OffsetsInitializer startingOffsetsInitializer; + private final OffsetsInitializer stoppingOffsetsInitializer; + private final BucketOffsetsRetriever bucketOffsetsRetriever; + private final Supplier> partitionSupplier; + private final KvSnapshotsRetriever kvSnapshotsRetriever; + private final Predicate tableBucketSkipper; + + FlussOnlyBatchSplitGenerator( + TableInfo tableInfo, + boolean hasPrimaryKey, + boolean isPartitioned, + OffsetsInitializer startingOffsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + BucketOffsetsRetriever bucketOffsetsRetriever, + Supplier> partitionSupplier, + KvSnapshotsRetriever kvSnapshotsRetriever, + Predicate tableBucketSkipper) { + this.tableInfo = tableInfo; + this.hasPrimaryKey = hasPrimaryKey; + this.isPartitioned = isPartitioned; + this.startingOffsetsInitializer = startingOffsetsInitializer; + this.stoppingOffsetsInitializer = stoppingOffsetsInitializer; + this.bucketOffsetsRetriever = bucketOffsetsRetriever; + this.partitionSupplier = partitionSupplier; + this.kvSnapshotsRetriever = kvSnapshotsRetriever; + this.tableBucketSkipper = tableBucketSkipper; + } + + List generate() { + if (isPartitioned) { + Set partitionInfos = partitionSupplier.get(); + return hasPrimaryKey + ? generatePrimaryKeyTableSplits(partitionInfos) + : generateLogTableSplits(partitionInfos); + } else { + return hasPrimaryKey + ? getBatchSnapshotAndLogSplits(kvSnapshotsRetriever.get(null), null) + : getLogSplits(null, null); + } + } + + private List generatePrimaryKeyTableSplits( + Collection partitions) { + List splits = new ArrayList<>(); + for (PartitionInfo partition : partitions) { + String partitionName = partition.getPartitionName(); + splits.addAll( + getBatchSnapshotAndLogSplits( + kvSnapshotsRetriever.get(partitionName), partitionName)); + } + return splits; + } + + private List generateLogTableSplits(Collection partitions) { + List splits = new ArrayList<>(); + for (PartitionInfo partition : partitions) { + splits.addAll(getLogSplits(partition.getPartitionId(), partition.getPartitionName())); + } + return splits; + } + + private List getBatchSnapshotAndLogSplits( + KvSnapshots snapshots, @Nullable String partitionName) { + long tableId = snapshots.getTableId(); + Long partitionId = snapshots.getPartitionId(); + List splits = new ArrayList<>(); + List bucketsNeedInitOffset = new ArrayList<>(); + for (Integer bucketId : snapshots.getBucketIds()) { + TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId); + if (!tableBucketSkipper.test(tableBucket)) { + bucketsNeedInitOffset.add(bucketId); + } + } + + Map stoppingOffsets = + bucketsNeedInitOffset.isEmpty() + ? Collections.emptyMap() + : stoppingOffsetsInitializer.getBucketOffsets( + partitionName, bucketsNeedInitOffset, bucketOffsetsRetriever); + + for (Integer bucketId : bucketsNeedInitOffset) { + TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId); + OptionalLong snapshotId = snapshots.getSnapshotId(bucketId); + long batchSnapshotId = + snapshotId.isPresent() + ? snapshotId.getAsLong() + : HybridSnapshotLogSplit.NO_SNAPSHOT_ID; + long logStartingOffset; + if (snapshotId.isPresent()) { + OptionalLong logOffset = snapshots.getLogOffset(bucketId); + checkState( + logOffset.isPresent(), + "Log offset should be present if snapshot id is present."); + logStartingOffset = logOffset.getAsLong(); + } else { + logStartingOffset = LogScanner.EARLIEST_OFFSET; + } + + Long logStoppingOffset = stoppingOffsets.get(bucketId); + checkState( + logStoppingOffset != null, + "Stopping offset should be present for bucket %s.", + bucketId); + splits.add( + new HybridSnapshotLogSplit( + tableBucket, + partitionName, + batchSnapshotId, + 0, + false, + logStartingOffset, + logStoppingOffset, + true)); + } + return splits; + } + + private List getLogSplits( + @Nullable Long partitionId, @Nullable String partitionName) { + List splits = new ArrayList<>(); + List bucketsNeedInitOffset = new ArrayList<>(); + for (int bucketId = 0; bucketId < tableInfo.getNumBuckets(); bucketId++) { + TableBucket tableBucket = + new TableBucket(tableInfo.getTableId(), partitionId, bucketId); + if (!tableBucketSkipper.test(tableBucket)) { + bucketsNeedInitOffset.add(bucketId); + } + } + + if (!bucketsNeedInitOffset.isEmpty()) { + Map startingOffsets = + startingOffsetsInitializer.getBucketOffsets( + partitionName, bucketsNeedInitOffset, bucketOffsetsRetriever); + Map stoppingOffsets = + stoppingOffsetsInitializer.getBucketOffsets( + partitionName, bucketsNeedInitOffset, bucketOffsetsRetriever); + for (Integer bucketId : bucketsNeedInitOffset) { + Long startingOffset = startingOffsets.get(bucketId); + Long stoppingOffset = stoppingOffsets.get(bucketId); + checkState( + startingOffset != null, + "Starting offset should be present for bucket %s.", + bucketId); + checkState( + stoppingOffset != null, + "Stopping offset should be present for bucket %s.", + bucketId); + splits.add( + new LogSplit( + new TableBucket(tableInfo.getTableId(), partitionId, bucketId), + partitionName, + startingOffset, + stoppingOffset)); + } + } + return splits; + } + + /** Retrieves latest KV snapshots and keeps any required snapshot leases alive. */ + @FunctionalInterface + interface KvSnapshotsRetriever { + KvSnapshots get(@Nullable String partitionName); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java index 90f2b6e9a6..6c9648284a 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java @@ -23,6 +23,7 @@ import org.apache.fluss.client.table.Table; import org.apache.fluss.client.table.scanner.ScanRecord; import org.apache.fluss.client.table.scanner.batch.BatchScanner; +import org.apache.fluss.client.table.scanner.batch.KvSnapshotAndLogBatchScanner; import org.apache.fluss.client.table.scanner.log.LogScanner; import org.apache.fluss.client.table.scanner.log.ScanRecords; import org.apache.fluss.config.Configuration; @@ -208,12 +209,16 @@ public void handleSplitsChanges(SplitsChange splitsChanges) { HybridSnapshotLogSplit hybridSnapshotLogSplit = sourceSplitBase.asHybridSnapshotLogSplit(); - // if snapshot is not finished, add to pending snapshot splits - if (!hybridSnapshotLogSplit.isSnapshotFinished()) { + if (hybridSnapshotLogSplit.isBatch()) { boundedSplits.add(sourceSplitBase); + } else { + // if snapshot is not finished, add to pending snapshot splits + if (!hybridSnapshotLogSplit.isSnapshotFinished()) { + boundedSplits.add(sourceSplitBase); + } + // still need to subscribe log + subscribeLog(sourceSplitBase, hybridSnapshotLogSplit.getLogStartingOffset()); } - // still need to subscribe log - subscribeLog(sourceSplitBase, hybridSnapshotLogSplit.getLogStartingOffset()); } else if (sourceSplitBase.isLogSplit()) { subscribeLog(sourceSplitBase, sourceSplitBase.asLogSplit().getStartingOffset()); } else if (sourceSplitBase.isLakeSplit()) { @@ -404,14 +409,38 @@ private void checkSnapshotSplitOrStartNext() { // start to read next snapshot split currentBoundedSplit = nextSplit; if (currentBoundedSplit.isHybridSnapshotLogSplit()) { - SnapshotSplit snapshotSplit = currentBoundedSplit.asHybridSnapshotLogSplit(); - BatchScanner batchScanner = - table.newScan() - .project(projectedFields) - .createBatchScanner( - snapshotSplit.getTableBucket(), snapshotSplit.getSnapshotId()); - currentBoundedSplitReader = - new BoundedSplitReader(batchScanner, snapshotSplit.recordsToSkip()); + HybridSnapshotLogSplit hybridSnapshotLogSplit = + currentBoundedSplit.asHybridSnapshotLogSplit(); + if (hybridSnapshotLogSplit.isBatch()) { + BatchScanner batchScanner = + new KvSnapshotAndLogBatchScanner( + table, + hybridSnapshotLogSplit.getTableBucket(), + hybridSnapshotLogSplit.getSnapshotId(), + hybridSnapshotLogSplit.getLogStartingOffset(), + hybridSnapshotLogSplit + .getLogStoppingOffset() + .orElseThrow( + () -> + new IllegalStateException( + "Batch hybrid snapshot log split " + + "must have a stopping " + + "offset.")), + projectedFields); + currentBoundedSplitReader = + new BoundedSplitReader( + batchScanner, hybridSnapshotLogSplit.recordsToSkip()); + } else { + SnapshotSplit snapshotSplit = currentBoundedSplit.asHybridSnapshotLogSplit(); + BatchScanner batchScanner = + table.newScan() + .project(projectedFields) + .createBatchScanner( + snapshotSplit.getTableBucket(), + snapshotSplit.getSnapshotId()); + currentBoundedSplitReader = + new BoundedSplitReader(batchScanner, snapshotSplit.recordsToSkip()); + } } else if (currentBoundedSplit.isLakeSplit()) { currentBoundedSplitReader = getLakeSplitReader().getBoundedSplitScanner(currentBoundedSplit); @@ -530,7 +559,9 @@ private long getStoppingOffset(TableBucket tableBucket) { private FlinkRecordsWithSplitIds finishCurrentBoundedSplit() throws IOException { Set finishedSplits = - currentBoundedSplit instanceof HybridSnapshotLogSplit + (currentBoundedSplit instanceof HybridSnapshotLogSplit + && !((HybridSnapshotLogSplit) currentBoundedSplit) + .isBatch()) || (currentBoundedSplit instanceof LakeSnapshotAndFlussLogSplit && ((LakeSnapshotAndFlussLogSplit) currentBoundedSplit) .isStreaming()) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplit.java index 5c8f30e282..b2566de8de 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplit.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplit.java @@ -22,6 +22,9 @@ import javax.annotation.Nullable; import java.util.Objects; +import java.util.Optional; + +import static org.apache.fluss.utils.Preconditions.checkArgument; /** * The hybrid split for first reading the snapshot files and then switch to read the cdc log from a @@ -32,17 +35,13 @@ */ public class HybridSnapshotLogSplit extends SnapshotSplit { + public static final long NO_SNAPSHOT_ID = -1L; + private static final String HYBRID_SPLIT_PREFIX = "hybrid-snapshot-log-"; private final boolean isSnapshotFinished; private final long logStartingOffset; - - public HybridSnapshotLogSplit( - TableBucket tableBucket, - @Nullable String partitionName, - long snapshotId, - long logStartingOffset) { - this(tableBucket, partitionName, snapshotId, 0, false, logStartingOffset); - } + private final long logStoppingOffset; + private final boolean isBatch; public HybridSnapshotLogSplit( TableBucket tableBucket, @@ -50,16 +49,31 @@ public HybridSnapshotLogSplit( long snapshotId, long recordsToSkip, boolean isSnapshotFinished, - long logStartingOffset) { + long logStartingOffset, + long logStoppingOffset, + boolean isBatch) { super(tableBucket, partitionName, snapshotId, recordsToSkip); + checkArgument( + !isBatch || logStoppingOffset >= 0, + "Batch hybrid snapshot log split must have a non-negative stopping offset."); this.isSnapshotFinished = isSnapshotFinished; this.logStartingOffset = logStartingOffset; + this.logStoppingOffset = logStoppingOffset; + this.isBatch = isBatch; } public long getLogStartingOffset() { return logStartingOffset; } + public Optional getLogStoppingOffset() { + return logStoppingOffset >= 0 ? Optional.of(logStoppingOffset) : Optional.empty(); + } + + public boolean isBatch() { + return isBatch; + } + public boolean isSnapshotFinished() { return isSnapshotFinished; } @@ -82,12 +96,19 @@ public boolean equals(Object o) { } HybridSnapshotLogSplit that = (HybridSnapshotLogSplit) o; return isSnapshotFinished == that.isSnapshotFinished - && logStartingOffset == that.logStartingOffset; + && logStartingOffset == that.logStartingOffset + && logStoppingOffset == that.logStoppingOffset + && isBatch == that.isBatch; } @Override public int hashCode() { - return Objects.hash(super.hashCode(), isSnapshotFinished, logStartingOffset); + return Objects.hash( + super.hashCode(), + isSnapshotFinished, + logStartingOffset, + logStoppingOffset, + isBatch); } @Override @@ -103,6 +124,10 @@ public String toString() { + isSnapshotFinished + ", logStartingOffset=" + logStartingOffset + + ", logStoppingOffset=" + + logStoppingOffset + + ", isBatch=" + + isBatch + ", recordsToSkip=" + recordsToSkip + '}'; diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java index 6958518b53..a28d45a042 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java @@ -44,7 +44,9 @@ public HybridSnapshotLogSplit toSourceSplit() { hybridSnapshotLogSplit.getSnapshotId(), recordsToSkip, snapshotFinished, - nextOffset); + nextOffset, + hybridSnapshotLogSplit.getLogStoppingOffset().orElse(LogSplit.NO_STOPPING_OFFSET), + hybridSnapshotLogSplit.isBatch()); } public void setRecordsToSkip(long recordsToSkip) { diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/SourceSplitSerializer.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/SourceSplitSerializer.java index 7ee92cef9e..96275c76b3 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/SourceSplitSerializer.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/SourceSplitSerializer.java @@ -36,6 +36,7 @@ public class SourceSplitSerializer implements SimpleVersionedSerializer { private static final int VERSION_0 = 0; + private static final int VERSION_1 = 1; private static final ThreadLocal SERIALIZER_CACHE = ThreadLocal.withInitial(() -> new DataOutputSerializer(64)); @@ -43,7 +44,7 @@ public class SourceSplitSerializer implements SimpleVersionedSerializer lakeSource; @@ -75,6 +76,13 @@ public byte[] serialize(SourceSplitBase split) throws IOException { out.writeBoolean(hybridSnapshotLogSplit.isSnapshotFinished()); // write log starting offset out.writeLong(hybridSnapshotLogSplit.getLogStartingOffset()); + // write log stopping offset + out.writeLong( + hybridSnapshotLogSplit + .getLogStoppingOffset() + .orElse(LogSplit.NO_STOPPING_OFFSET)); + // write whether this is a bounded batch split + out.writeBoolean(hybridSnapshotLogSplit.isBatch()); } else { LogSplit logSplit = split.asLogSplit(); // write starting offset @@ -111,7 +119,7 @@ private void serializeSourceSplitBase(DataOutputSerializer out, SourceSplitBase @Override public SourceSplitBase deserialize(int version, byte[] serialized) throws IOException { - if (version != VERSION_0) { + if (version != VERSION_0 && version != VERSION_1) { throw new IOException("Unknown version " + version); } final DataInputDeserializer in = new DataInputDeserializer(serialized); @@ -133,13 +141,21 @@ public SourceSplitBase deserialize(int version, byte[] serialized) throws IOExce long recordsToSkip = in.readLong(); boolean isSnapshotFinished = in.readBoolean(); long logStartingOffset = in.readLong(); + long logStoppingOffset = LogSplit.NO_STOPPING_OFFSET; + boolean isBatch = false; + if (version == VERSION_1) { + logStoppingOffset = in.readLong(); + isBatch = in.readBoolean(); + } return new HybridSnapshotLogSplit( tableBucket, partitionName, snapshotId, recordsToSkip, isSnapshotFinished, - logStartingOffset); + logStartingOffset, + logStoppingOffset, + isBatch); } else if (splitKind == LOG_SPLIT_FLAG) { long startingOffset = in.readLong(); long stoppingOffset = in.readLong(); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java index 51d05fa036..e1337558b4 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java @@ -46,6 +46,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder; @@ -242,16 +243,102 @@ void testFilterOnLookupSource() throws Exception { } @Test - void testScanSingleRowFilterException() throws Exception { + void testScanWithIncompletePrimaryKeyFilter() throws Exception { String tableName = prepareSourceTable(new String[] {"id", "name"}, null); String query = String.format("SELECT * FROM %s WHERE id = 1", tableName); - // doesn't have all condition for primary key, doesn't support to execute - assertThatThrownBy(() -> tEnv.explainSql(query)) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessage( - "Currently, Fluss only support queries on table with datalake enabled" - + " or point queries on primary key when it's in batch execution mode."); + CloseableIterator collected = tEnv.executeSql(query).collect(); + List expected = Collections.singletonList("+I[1, address1, name1]"); + assertResultsIgnoreOrder(collected, expected, true); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testScanFullPrimaryKeyTable(boolean partitionTable) throws Exception { + String tableName = + partitionTable + ? prepareSourceTable(new String[] {"id", "dt"}, "dt") + : prepareSourceTable(new String[] {"id", "name"}, null); + String query = String.format("SELECT * FROM %s", tableName); + + CloseableIterator collected = tEnv.executeSql(query).collect(); + List expected; + if (partitionTable) { + String partition = + waitUntilPartitions( + FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), + TablePath.of(DEFAULT_DB, tableName)) + .values() + .stream() + .sorted() + .findFirst() + .get(); + expected = + Arrays.asList( + String.format("+I[1, address1, name1, %s]", partition), + String.format("+I[2, address2, name2, %s]", partition), + String.format("+I[3, address3, name3, %s]", partition), + String.format("+I[4, address4, name4, %s]", partition), + String.format("+I[5, address5, name5, %s]", partition)); + } else { + expected = + Arrays.asList( + "+I[1, address1, name1]", + "+I[2, address2, name2]", + "+I[3, address3, name3]", + "+I[4, address4, name4]", + "+I[5, address5, name5]"); + } + assertResultsIgnoreOrder(collected, expected, true); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testScanFullLogTable(boolean partitionTable) throws Exception { + String tableName = partitionTable ? preparePartitionedLogTable() : prepareLogTable(); + String query = String.format("SELECT * FROM %s", tableName); + + CloseableIterator collected = tEnv.executeSql(query).collect(); + List expected; + if (partitionTable) { + Collection partitions = + waitUntilPartitions( + FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), + TablePath.of(DEFAULT_DB, tableName)) + .values(); + expected = + partitions.stream() + .flatMap( + partition -> + Arrays.stream( + new String[] { + String.format( + "+I[1, address1, name1, %s]", + partition), + String.format( + "+I[2, null, name2, %s]", + partition), + String.format( + "+I[3, address3, name3, %s]", + partition), + String.format( + "+I[4, null, name4, %s]", + partition), + String.format( + "+I[5, address5, name5, %s]", + partition) + })) + .collect(Collectors.toList()); + } else { + expected = + Arrays.asList( + "+I[1, address1, name1]", + "+I[2, null, name2]", + "+I[3, address3, name3]", + "+I[4, null, name4]", + "+I[5, address5, name5]"); + } + assertResultsIgnoreOrder(collected, expected, true); } @Test @@ -313,6 +400,60 @@ void testLimitPrimaryTableScan() throws Exception { .hasMessageContaining("LIMIT statement doesn't support greater than 2048"); } + @Test + void testPrimaryKeyTableBatchScanMergesSnapshotAndLog() throws Exception { + String tableName = String.format("test_pk_batch_snapshot_log_%s", RandomUtils.nextInt()); + tEnv.executeSql( + String.format( + "create table %s (" + + " id int not null," + + " address varchar," + + " name varchar," + + " primary key (id) NOT ENFORCED)" + + " with ('bucket.num' = '4')", + tableName)); + + TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); + try (Table table = conn.getTable(tablePath)) { + UpsertWriter upsertWriter = table.newUpsert().createWriter(); + upsertWriter.upsert(row(1, "address1", "name1")); + upsertWriter.upsert(row(2, "address2", "name2")); + upsertWriter.upsert(row(3, "address3", "name3")); + upsertWriter.flush(); + + FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tablePath); + + upsertWriter.upsert(row(1, "address11", "name11")); + upsertWriter.delete(row(2, null, null)); + upsertWriter.upsert(row(4, "address4", "name4")); + upsertWriter.flush(); + } + + CloseableIterator collected = + tEnv.executeSql(String.format("SELECT * FROM %s", tableName)).collect(); + List expected = + Arrays.asList( + "+I[1, address11, name11]", + "+I[3, address3, name3]", + "+I[4, address4, name4]"); + assertResultsIgnoreOrder(collected, expected, true); + } + + @Test + void testPrimaryKeyTableBatchScanRejectsNonFullStartupMode() throws Exception { + String tableName = prepareSourceTable(new String[] {"id"}, null); + String query = + String.format( + "SELECT * FROM %s /*+ OPTIONS('scan.startup.mode' = 'earliest') */", + tableName); + + assertThatThrownBy(() -> tEnv.explainSql(query)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage( + "Currently, Fluss batch scan on primary-key tables only supports " + + "full startup mode."); + } + @Test void testLimitLogTableScan() throws Exception { String tableName = prepareLogTable(); @@ -354,6 +495,20 @@ void testLimitLogTableScan() throws Exception { assertThat(collected).hasSize(3); } + @Test + void testLogTableBatchScanSupportsNonFullStartupMode() throws Exception { + String tableName = prepareLogTable(); + String query = + String.format( + "SELECT COUNT(address) FROM %s " + + "/*+ OPTIONS('scan.startup.mode' = 'earliest') */", + tableName); + + CloseableIterator iterRows = tEnv.executeSql(query).collect(); + List collected = collectRowsWithTimeout(iterRows, 1); + assertThat(collected).isEqualTo(Collections.singletonList("+I[3]")); + } + @Test void testLimitLogTableScanWithComplexTypes() throws Exception { String tableName = prepareLogTableWithComplexTypes(); @@ -414,33 +569,21 @@ void testCountPushDownForPkTable(boolean partitionTable) throws Exception { assertThat(collected).isEqualTo(expected); // test COUNT(column) on nullable column - should NOT push down - // For PK table, this will fail because it doesn't support full scan in batch mode - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format("SELECT COUNT(address) FROM %s", tableName))) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + query = String.format("SELECT COUNT(address) FROM %s", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 1); + assertThat(collected).isEqualTo(expected); - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format( - "SELECT COUNT(DISTINCT address) FROM %s", - tableName))) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + query = String.format("SELECT COUNT(DISTINCT address) FROM %s", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 1); + assertThat(collected).isEqualTo(expected); // test not push down grouping count. - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format( - "SELECT COUNT(*) FROM %s group by id", - tableName)) - .wait()) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + query = String.format("SELECT COUNT(*) FROM %s group by id", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 5); + assertThat(collected).containsOnly("+I[1]"); } @Test @@ -489,32 +632,23 @@ void testCountPushDownForLogTable(boolean partitionTable) throws Exception { assertThat(collected).isEqualTo(expected); // test COUNT(column) with NULL values - should NOT push down for nullable columns - // This will fail because log table doesn't support full scan in batch mode - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format("SELECT COUNT(address) FROM %s", tableName))) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format( - "SELECT COUNT(DISTINCT address) FROM %s", - tableName))) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + query = String.format("SELECT COUNT(address) FROM %s", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 1); + assertThat(collected) + .isEqualTo( + Collections.singletonList(String.format("+I[%s]", partitionTable ? 6 : 3))); + + query = String.format("SELECT COUNT(DISTINCT address) FROM %s", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 1); + assertThat(collected).isEqualTo(Collections.singletonList("+I[3]")); // test not push down grouping count. - assertThatThrownBy( - () -> - tEnv.explainSql( - String.format( - "SELECT COUNT(*) FROM %s group by id", - tableName)) - .wait()) - .hasMessageContaining( - "Currently, Fluss only support queries on table with datalake enabled or point queries on primary key when it's in batch execution mode."); + query = String.format("SELECT COUNT(*) FROM %s group by id", tableName); + iterRows = tEnv.executeSql(query).collect(); + collected = collectRowsWithTimeout(iterRows, 5); + assertThat(collected).containsOnly(String.format("+I[%s]", partitionTable ? 2 : 1)); } private String prepareSourceTable(String[] keys, String partitionedKey) throws Exception { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitterTest.java index d2a647d041..6d28dfbcfd 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitterTest.java @@ -23,6 +23,7 @@ import org.apache.fluss.flink.source.reader.RecordAndPos; import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit; import org.apache.fluss.flink.source.split.HybridSnapshotLogSplitState; +import org.apache.fluss.flink.source.split.LogSplit; import org.apache.fluss.flink.source.testutils.Order; import org.apache.fluss.flink.source.testutils.OrderDeserializationSchema; import org.apache.fluss.flink.utils.FlussRowToFlinkRowConverter; @@ -54,7 +55,8 @@ void testEmitRowDataRecordWithHybridSplitInSnapshotPhase() throws Exception { TableBucket bucket0 = new TableBucket(tableId, 0); HybridSnapshotLogSplit hybridSnapshotLogSplit = - new HybridSnapshotLogSplit(bucket0, null, 0L, 0L); + new HybridSnapshotLogSplit( + bucket0, null, 0L, 0, false, 0L, LogSplit.NO_STOPPING_OFFSET, false); HybridSnapshotLogSplitState splitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); @@ -106,7 +108,8 @@ void testEmitPojoRecordWithHybridSplitInSnapshotPhase() throws Exception { TableBucket bucket0 = new TableBucket(tableId, 0); HybridSnapshotLogSplit hybridSnapshotLogSplit = - new HybridSnapshotLogSplit(bucket0, null, 0L, 0L); + new HybridSnapshotLogSplit( + bucket0, null, 0L, 0, false, 0L, LogSplit.NO_STOPPING_OFFSET, false); HybridSnapshotLogSplitState splitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java index 3f90a4b1b4..0c73afc3c2 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java @@ -356,21 +356,111 @@ void testPkTableWithSnapshotSplits() throws Throwable { 0, Collections.singletonList( new HybridSnapshotLogSplit( - bucket0, null, 0L, bucketIdToNumRecords.get(0)))); + bucket0, + null, + 0L, + 0, + false, + bucketIdToNumRecords.get(0), + LogSplit.NO_STOPPING_OFFSET, + false))); expectedAssignment.put( 1, Collections.singletonList( new HybridSnapshotLogSplit( - bucket1, null, 0L, bucketIdToNumRecords.get(1)))); + bucket1, + null, + 0L, + 0, + false, + bucketIdToNumRecords.get(1), + LogSplit.NO_STOPPING_OFFSET, + false))); expectedAssignment.put( 2, Collections.singletonList( new HybridSnapshotLogSplit( - bucket2, null, 0L, bucketIdToNumRecords.get(2)))); + bucket2, + null, + 0L, + 0, + false, + bucketIdToNumRecords.get(2), + LogSplit.NO_STOPPING_OFFSET, + false))); checkSplitAssignmentIgnoreSnapshotFiles(expectedAssignment, actualAssignment); } } + @Test + void testBatchPkTableWithSnapshotSplits() throws Throwable { + createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + int numSubtasks = 5; + putRows(DEFAULT_TABLE_PATH, 10); + FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(DEFAULT_TABLE_PATH); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(numSubtasks); + FlinkSourceEnumerator enumerator = + new FlinkSourceEnumerator( + DEFAULT_TABLE_PATH, + flussConf, + true, + false, + context, + OffsetsInitializer.full(), + DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS, + false, + null, + null, + LeaseContext.DEFAULT, + false)) { + enumerator.start(); + for (int i = 0; i < numSubtasks; i++) { + registerReader(context, enumerator, i); + } + context.runNextOneTimeCallable(); + + List assignedSplits = + getReadersAssignments(context).values().stream() + .flatMap(List::stream) + .collect(Collectors.toList()); + assertThat(assignedSplits).hasSize(DEFAULT_BUCKET_NUM); + assertThat(assignedSplits) + .allSatisfy( + split -> { + assertThat(split).isInstanceOf(HybridSnapshotLogSplit.class); + assertThat(split.asHybridSnapshotLogSplit().isBatch()).isTrue(); + }); + } + } + + @Test + void testBatchPkTableRejectsNonFullStartupMode() throws Throwable { + createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(1); + FlinkSourceEnumerator enumerator = + new FlinkSourceEnumerator( + DEFAULT_TABLE_PATH, + flussConf, + true, + false, + context, + OffsetsInitializer.earliest(), + DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS, + false, + null, + null, + LeaseContext.DEFAULT, + false)) { + assertThatThrownBy(enumerator::start) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage( + "Batch mode on primary-key tables only supports full startup mode."); + } + } + @Test void testNonPkTable() throws Throwable { int numSubtasks = 3; @@ -939,13 +1029,17 @@ void testGetSplitOwner() throws Exception { // test splits for same non-partitioned bucket, should assign to same task TableBucket t1 = new TableBucket(tableId, 0); SourceSplitBase s1 = new LogSplit(t1, null, 1); - SourceSplitBase s2 = new HybridSnapshotLogSplit(t1, null, 0L, 1); + SourceSplitBase s2 = + new HybridSnapshotLogSplit( + t1, null, 0L, 0, false, 1, LogSplit.NO_STOPPING_OFFSET, false); assertThat(enumerator.getSplitOwner(s1)).isEqualTo(enumerator.getSplitOwner(s2)); // test splits for same partitioned bucket, should assign to same task t1 = new TableBucket(tableId, 1L, 0); s1 = new LogSplit(t1, "p1", 1); - s2 = new HybridSnapshotLogSplit(t1, "p1", 0L, 2); + s2 = + new HybridSnapshotLogSplit( + t1, "p1", 0L, 0, false, 2, LogSplit.NO_STOPPING_OFFSET, false); assertThat(enumerator.getSplitOwner(s1)).isEqualTo(enumerator.getSplitOwner(s2)); // test splits for partitioned bucket diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java index c170518995..5b0117a0f7 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java @@ -476,7 +476,9 @@ private static String toLogSplitId(TableBucket tableBucket) { private static String toHybridSnapshotLogSplitId(TableBucket tableBucket) { // snapshotId and logOffset doesn't affect splitId, use mocked 0 value. - return new HybridSnapshotLogSplit(tableBucket, null, 0, 0).splitId(); + return new HybridSnapshotLogSplit( + tableBucket, null, 0, 0, false, 0, LogSplit.NO_STOPPING_OFFSET, false) + .splitId(); } private static int getBucketId(InternalRow row) { @@ -499,7 +501,14 @@ private List getHybridSnapshotLogSplits(TablePath tablePath) th if (snapshotId.isPresent() && logOffset.isPresent()) { hybridSnapshotLogSplits.add( new HybridSnapshotLogSplit( - tableBucket, null, snapshotId.getAsLong(), logOffset.getAsLong())); + tableBucket, + null, + snapshotId.getAsLong(), + 0, + false, + logOffset.getAsLong(), + LogSplit.NO_STOPPING_OFFSET, + false)); } } return hybridSnapshotLogSplits; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/split/SourceSplitSerializerTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/split/SourceSplitSerializerTest.java index 54a3207807..33cefaaf6c 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/split/SourceSplitSerializerTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/split/SourceSplitSerializerTest.java @@ -19,6 +19,7 @@ import org.apache.fluss.metadata.TableBucket; +import org.apache.flink.core.memory.DataOutputSerializer; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -39,12 +40,21 @@ class SourceSplitSerializerTest { @ValueSource(booleans = {true, false}) void testHybridSnapshotLogSplitSerde(boolean isPartitioned) throws Exception { int snapshotId = 100; - int recordsToSkip = 3; + long logStartingOffset = 3L; + long recordsToSkip = 4L; TableBucket bucket = isPartitioned ? partitionedTableBucket : tableBucket; String partitionName = isPartitioned ? "2024" : null; HybridSnapshotLogSplit split = - new HybridSnapshotLogSplit(bucket, partitionName, snapshotId, recordsToSkip); + new HybridSnapshotLogSplit( + bucket, + partitionName, + snapshotId, + 0, + false, + logStartingOffset, + LogSplit.NO_STOPPING_OFFSET, + false); byte[] serialized = serializer.serialize(split); SourceSplitBase deserializedSplit = @@ -53,7 +63,14 @@ void testHybridSnapshotLogSplitSerde(boolean isPartitioned) throws Exception { split = new HybridSnapshotLogSplit( - bucket, partitionName, snapshotId, recordsToSkip, true, 5); + bucket, + partitionName, + snapshotId, + recordsToSkip, + true, + 5, + LogSplit.NO_STOPPING_OFFSET, + false); serialized = serializer.serialize(split); deserializedSplit = serializer.deserialize(serializer.getVersion(), serialized); assertThat(deserializedSplit).isEqualTo(split); @@ -71,4 +88,56 @@ void testLogSplitSerde(boolean isPartitioned) throws Exception { serializer.deserialize(serializer.getVersion(), serialized); assertThat(deserializedSplit).isEqualTo(logSplit); } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testHybridSnapshotLogSplitSerdeWithStoppingOffset(boolean isPartitioned) throws Exception { + TableBucket bucket = isPartitioned ? partitionedTableBucket : tableBucket; + String partitionName = isPartitioned ? "2024" : null; + HybridSnapshotLogSplit split = + new HybridSnapshotLogSplit(bucket, partitionName, 100L, 0L, false, 20L, 30L, true); + + byte[] serialized = serializer.serialize(split); + SourceSplitBase deserializedSplit = + serializer.deserialize(serializer.getVersion(), serialized); + + assertThat(deserializedSplit).isEqualTo(split); + assertThat(deserializedSplit.asHybridSnapshotLogSplit().isBatch()).isTrue(); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testDeserializeVersionZeroHybridSnapshotLogSplit(boolean isPartitioned) throws Exception { + TableBucket bucket = isPartitioned ? partitionedTableBucket : tableBucket; + String partitionName = isPartitioned ? "2024" : null; + DataOutputSerializer out = new DataOutputSerializer(64); + out.writeByte(SourceSplitBase.HYBRID_SNAPSHOT_SPLIT_FLAG); + out.writeLong(bucket.getTableId()); + out.writeBoolean(bucket.getPartitionId() != null); + if (bucket.getPartitionId() != null) { + out.writeLong(bucket.getPartitionId()); + out.writeUTF(partitionName); + } + out.writeInt(bucket.getBucket()); + out.writeLong(100L); + out.writeLong(3L); + out.writeBoolean(false); + out.writeLong(20L); + + SourceSplitBase deserializedSplit = serializer.deserialize(0, out.getCopyOfBuffer()); + + assertThat(deserializedSplit) + .isEqualTo( + new HybridSnapshotLogSplit( + bucket, + partitionName, + 100L, + 3L, + false, + 20L, + LogSplit.NO_STOPPING_OFFSET, + false)); + assertThat(deserializedSplit.asHybridSnapshotLogSplit().isBatch()).isFalse(); + assertThat(deserializedSplit.asHybridSnapshotLogSplit().getLogStoppingOffset()).isEmpty(); + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java index 7394eddea5..4c4523d83d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceEnumeratorStateSerializerTest.java @@ -67,7 +67,15 @@ void testPendingSplitsCheckpointSerde() throws Exception { // Add a HybridSnapshotLogSplit TableBucket hybridSplitBucket = new TableBucket(1, 1); HybridSnapshotLogSplit hybridSplit = - new HybridSnapshotLogSplit(hybridSplitBucket, null, 200L, 50L); + new HybridSnapshotLogSplit( + hybridSplitBucket, + null, + 200L, + 0, + false, + 50L, + LogSplit.NO_STOPPING_OFFSET, + false); remainingHybridLakeFlussSplits.add(hybridSplit); // Add a LakeSnapshotAndFlussLogSplit @@ -232,7 +240,16 @@ void testInitialDiscoveryFinished() throws Exception { List unassignedSplits = new ArrayList<>(); unassignedSplits.add(new LogSplit(new TableBucket(1, 0), null, 100L)); unassignedSplits.add(new LogSplit(new TableBucket(1, 5L, 1), "p5", 200L)); - unassignedSplits.add(new HybridSnapshotLogSplit(new TableBucket(1, 2), null, 300L, 50L)); + unassignedSplits.add( + new HybridSnapshotLogSplit( + new TableBucket(1, 2), + null, + 300L, + 0, + false, + 50L, + LogSplit.NO_STOPPING_OFFSET, + false)); SourceEnumeratorState sourceEnumeratorState = new SourceEnumeratorState( diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceSplitStateTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceSplitStateTest.java index e122a13ee0..ec11d18831 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceSplitStateTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/state/SourceSplitStateTest.java @@ -61,7 +61,15 @@ void testHybridSnapshotLogSplitState() { // verify the split that snapshot is not finished HybridSnapshotLogSplit hybridSnapshotLogSplit = - new HybridSnapshotLogSplit(tableBucket, "partition1", 1L, 100L); + new HybridSnapshotLogSplit( + tableBucket, + "partition1", + 1L, + 0, + false, + 100L, + LogSplit.NO_STOPPING_OFFSET, + false); HybridSnapshotLogSplitState hybridSnapshotLogSplitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); assertThat(hybridSnapshotLogSplitState.toSourceSplit()).isEqualTo(hybridSnapshotLogSplit); @@ -69,20 +77,44 @@ void testHybridSnapshotLogSplitState() { // set record to skip and verify hybridSnapshotLogSplitState.setRecordsToSkip(200L); HybridSnapshotLogSplit expectedHybridSnapshotLogSplit = - new HybridSnapshotLogSplit(tableBucket, "partition1", 1L, 200L, false, 100L); + new HybridSnapshotLogSplit( + tableBucket, + "partition1", + 1L, + 200L, + false, + 100L, + LogSplit.NO_STOPPING_OFFSET, + false); assertThat(hybridSnapshotLogSplitState.toSourceSplit()) .isEqualTo(expectedHybridSnapshotLogSplit); // set next offset and verify hybridSnapshotLogSplitState.setNextOffset(500L); expectedHybridSnapshotLogSplit = - new HybridSnapshotLogSplit(tableBucket, "partition1", 1L, 200L, true, 500L); + new HybridSnapshotLogSplit( + tableBucket, + "partition1", + 1L, + 200L, + true, + 500L, + LogSplit.NO_STOPPING_OFFSET, + false); assertThat(hybridSnapshotLogSplitState.toSourceSplit()) .isEqualTo(expectedHybridSnapshotLogSplit); // verify the split that snapshot is finished hybridSnapshotLogSplit = - new HybridSnapshotLogSplit(tableBucket, "partition1", 1L, 100L, true, 100L); + new HybridSnapshotLogSplit( + tableBucket, + "partition1", + 1L, + 100L, + true, + 100L, + LogSplit.NO_STOPPING_OFFSET, + false); hybridSnapshotLogSplitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); assertThat(hybridSnapshotLogSplitState.toSourceSplit()).isEqualTo(hybridSnapshotLogSplit); @@ -90,7 +122,27 @@ void testHybridSnapshotLogSplitState() { hybridSnapshotLogSplitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); hybridSnapshotLogSplitState.setNextOffset(500L); expectedHybridSnapshotLogSplit = - new HybridSnapshotLogSplit(tableBucket, "partition1", 1L, 100L, true, 500L); + new HybridSnapshotLogSplit( + tableBucket, + "partition1", + 1L, + 100L, + true, + 500L, + LogSplit.NO_STOPPING_OFFSET, + false); + assertThat(hybridSnapshotLogSplitState.toSourceSplit()) + .isEqualTo(expectedHybridSnapshotLogSplit); + + // verify the bounded split keeps its stopping offset when state is materialized + hybridSnapshotLogSplit = + new HybridSnapshotLogSplit( + tableBucket, "partition1", 1L, 100L, false, 100L, 1000L, true); + hybridSnapshotLogSplitState = new HybridSnapshotLogSplitState(hybridSnapshotLogSplit); + hybridSnapshotLogSplitState.setRecordsToSkip(200L); + expectedHybridSnapshotLogSplit = + new HybridSnapshotLogSplit( + tableBucket, "partition1", 1L, 200L, false, 100L, 1000L, true); assertThat(hybridSnapshotLogSplitState.toSourceSplit()) .isEqualTo(expectedHybridSnapshotLogSplit); } diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadDataStreamITCase.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadDataStreamITCase.java index 6fe9ef3fff..f5dd98ea85 100644 --- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadDataStreamITCase.java +++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkUnionReadDataStreamITCase.java @@ -238,7 +238,7 @@ void testUnionReadWithFilterPushdown() throws Exception { void testBoundedRequiresFullStartupModeFailsFast() throws Exception { String tableName = "ds_bounded_invalid_startup"; TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); - createLogTable(tablePath, false); + createPkTable(tablePath, false); FlussSourceBuilder builder = FlussSource.builder() @@ -251,27 +251,8 @@ void testBoundedRequiresFullStartupModeFailsFast() throws Exception { assertThatThrownBy(builder::build) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Bounded (batch) read requires"); - } - - @Test - void testBoundedRequiresDataLakeEnabledFailsFast() throws Exception { - String tableName = "ds_bounded_no_lake"; - TablePath tablePath = TablePath.of(DEFAULT_DB, tableName); - createNonLakeLogTable(tablePath); - - FlussSourceBuilder builder = - FlussSource.builder() - .setBootstrapServers(bootstrapServers()) - .setDatabase(DEFAULT_DB) - .setTable(tableName) - .setStartingOffsets(OffsetsInitializer.full()) - .setBounded() - .setDeserializationSchema(new RowDataDeserializationSchema()); - - assertThatThrownBy(builder::build) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Bounded (batch) read requires"); + .hasMessageContaining( + "Bounded (batch) read on primary-key tables requires full mode"); } @Test