This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b2eba89c1b [flink] Expose scan.bucket for single-bucket manifest
pruning (#8117)
b2eba89c1b is described below
commit b2eba89c1b8a57e6613035a167fbacd0422fdab9
Author: wangwj <[email protected]>
AuthorDate: Fri Jul 3 20:22:02 2026 +0800
[flink] Expose scan.bucket for single-bucket manifest pruning (#8117)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 14 ++
.../paimon/table/AbstractFileStoreTable.java | 30 ++-
.../paimon/table/source/AbstractDataTableScan.java | 38 ++++
.../paimon/table/source/DataTableBatchScan.java | 6 +
.../apache/paimon/table/source/ScanBucketTest.java | 240 +++++++++++++++++++++
.../org/apache/paimon/flink/ScanBucketITCase.java | 174 +++++++++++++++
7 files changed, 499 insertions(+), 9 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index f75de298d6..e31bd111b1 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1319,6 +1319,12 @@ This config option does not affect the default
filesystem metastore.</td>
<td>Integer</td>
<td>The parallelism of scanning manifest files, default value is
the size of cpu processor. Note: Scale-up this parameter will increase memory
usage while scanning manifest files. We can consider downsize it when we
encounter an out of memory exception while scanning</td>
</tr>
+ <tr>
+ <td><h5>scan.bucket</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>Specify a single bucket to scan. This option filters manifest
entries and only plans splits for the given bucket. It is only supported for
fixed-bucket primary key tables (bucket > 0). It cannot be used with
postpone bucket tables.</td>
+ </tr>
<tr>
<td><h5>scan.max-splits-per-task</h5></td>
<td style="word-wrap: break-word;">10</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index ee1db88285..ce15c207ec 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -635,6 +635,16 @@ public class CoreOptions implements Serializable {
"Max split size should be cached for one task
while scanning. "
+ "If splits size cached in enumerator are
greater than tasks size multiply by this value, scanner will pause scanning.");
+ public static final ConfigOption<Integer> SCAN_BUCKET =
+ key("scan.bucket")
+ .intType()
+ .noDefaultValue()
+ .withDescription(
+ "Specify a single bucket to scan. This option
filters manifest entries "
+ + "and only plans splits for the given
bucket. It is only supported "
+ + "for fixed-bucket primary key tables
(bucket > 0). It cannot be used "
+ + "with postpone bucket tables.");
+
@Immutable
public static final ConfigOption<MergeEngine> MERGE_ENGINE =
key("merge-engine")
@@ -3546,6 +3556,10 @@ public class CoreOptions implements Serializable {
return options.get(SCAN_MANIFEST_PARALLELISM);
}
+ public Integer scanBucket() {
+ return options.get(SCAN_BUCKET);
+ }
+
public Duration streamingReadDelay() {
return options.get(STREAMING_READ_SNAPSHOT_DELAY);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index a35167d884..3154e87341 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -296,6 +296,11 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
coreOptions(),
newSnapshotReader(),
catalogEnvironment.tableQueryAuth(coreOptions()));
+ Integer scanBucket = coreOptions().scanBucket();
+ if (scanBucket != null) {
+ DataTableBatchScan.validateScanBucketOption(tableSchema,
coreOptions(), scanBucket);
+ scan.withBucket(scanBucket);
+ }
if (coreOptions().dataEvolutionEnabled()) {
return new DataEvolutionBatchScan(this, scan);
}
@@ -304,15 +309,22 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@Override
public StreamDataTableScan newStreamScan() {
- return new DataTableStreamScan(
- tableSchema,
- coreOptions(),
- newSnapshotReader(),
- snapshotManager(),
- changelogManager(),
- supportStreamingReadOverwrite(),
- catalogEnvironment.tableQueryAuth(coreOptions()),
- !tableSchema.primaryKeys().isEmpty());
+ DataTableStreamScan scan =
+ new DataTableStreamScan(
+ tableSchema,
+ coreOptions(),
+ newSnapshotReader(),
+ snapshotManager(),
+ changelogManager(),
+ supportStreamingReadOverwrite(),
+ catalogEnvironment.tableQueryAuth(coreOptions()),
+ !tableSchema.primaryKeys().isEmpty());
+ Integer scanBucket = coreOptions().scanBucket();
+ if (scanBucket != null) {
+ DataTableBatchScan.validateScanBucketOption(tableSchema,
coreOptions(), scanBucket);
+ scan.withBucket(scanBucket);
+ }
+ return scan;
}
protected abstract SplitGenerator splitGenerator();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index 4908a85b32..adec8a9175 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -31,6 +31,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.source.snapshot.CompactedStartingScanner;
import
org.apache.paimon.table.source.snapshot.ContinuousCompactorStartingScanner;
import
org.apache.paimon.table.source.snapshot.ContinuousFromSnapshotFullStartingScanner;
@@ -123,6 +124,43 @@ abstract class AbstractDataTableScan implements
DataTableScan {
return this;
}
+ /** Validates {@link CoreOptions#SCAN_BUCKET} for primary-key fixed-bucket
tables. */
+ static void validateScanBucketOption(TableSchema schema, CoreOptions
coreOptions, int bucket) {
+ checkArgument(
+ !schema.primaryKeys().isEmpty(),
+ "Bucket scan is only supported for primary key tables.");
+ checkArgument(
+ bucketModeFromOption(coreOptions.bucket()) ==
BucketMode.HASH_FIXED,
+ "Bucket scan is only supported for fixed-bucket tables, but
got bucket mode %s.",
+ bucketModeFromOption(coreOptions.bucket()));
+ validateFixedBucketRange(coreOptions, bucket);
+ }
+
+ private static void validateFixedBucketRange(CoreOptions coreOptions, int
bucket) {
+ checkArgument(bucket >= 0, "Bucket id must be non-negative, but is
%s.", bucket);
+ int numBuckets = coreOptions.bucket();
+ checkArgument(
+ numBuckets > 0,
+ "Bucket scan is only supported for tables with bucket > 0, but
got bucket %s.",
+ numBuckets);
+ checkArgument(
+ bucket < numBuckets,
+ "Bucket id %s must be less than table bucket number %s.",
+ bucket,
+ numBuckets);
+ }
+
+ private static BucketMode bucketModeFromOption(int bucketOption) {
+ switch (bucketOption) {
+ case -2:
+ return BucketMode.POSTPONE_MODE;
+ case -1:
+ return BucketMode.HASH_DYNAMIC;
+ default:
+ return BucketMode.HASH_FIXED;
+ }
+ }
+
@Override
public AbstractDataTableScan withBucketFilter(Filter<Integer>
bucketFilter) {
snapshotReader.withBucketFilter(bucketFilter);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
index 6928ad9f9b..5293fad4ff 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
@@ -52,6 +52,12 @@ public class DataTableBatchScan extends
AbstractDataTableScan {
private static final Logger LOG =
LoggerFactory.getLogger(DataTableBatchScan.class);
+ /** Validates {@link CoreOptions#SCAN_BUCKET} for primary-key fixed-bucket
tables. */
+ public static void validateScanBucketOption(
+ TableSchema schema, CoreOptions coreOptions, int bucket) {
+ AbstractDataTableScan.validateScanBucketOption(schema, coreOptions,
bucket);
+ }
+
private StartingScanner startingScanner;
private boolean hasNext;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/ScanBucketTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/ScanBucketTest.java
new file mode 100644
index 0000000000..5a56db2e64
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/ScanBucketTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.paimon.table.source;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link CoreOptions#SCAN_BUCKET}. */
+public class ScanBucketTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void testWithBucketAllowsAppendOnlyFixedBucketTable() throws
Exception {
+ FileStoreTable table = createAppendOnlyFixedBucketTable("4");
+ writeRows(table, 1, 2, 3, 4, 5, 6, 7, 8);
+
+
assertThat(extractBuckets(table.newScan().withBucket(0).plan().splits()))
+ .containsExactly(0);
+
assertThat(extractBuckets(table.newReadBuilder().withBucket(0).newScan().plan().splits()))
+ .containsExactly(0);
+ }
+
+ @Test
+ public void testScanBucketOptionRejectsOutOfRangeBucketId() throws
Exception {
+ FileStoreTable table = createTableWithScanBucket("4", true, "5");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Bucket id 5 must be less than table
bucket number 4");
+ }
+
+ @Test
+ public void testScanBucketOptionRejectsDynamicBucketTable() throws
Exception {
+ FileStoreTable table = createTableWithScanBucket("-1", true, "0");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("fixed-bucket tables");
+ }
+
+ @Test
+ public void testScanBucketOptionRejectsPostponeBucketTable() throws
Exception {
+ FileStoreTable table = createTableWithScanBucket("-2", true, "0");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("fixed-bucket tables");
+ }
+
+ @Test
+ public void testScanBucketOptionRejectsBucketUnawareTable() throws
Exception {
+ FileStoreTable table = createBucketUnawareTableWithScanBucket("0");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("primary key tables");
+ }
+
+ @Test
+ public void testScanBucketOptionRejectsTableWithoutPrimaryKey() throws
Exception {
+ FileStoreTable table = createAppendOnlyTableWithScanBucket("4", "0");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("primary key tables");
+ }
+
+ @Test
+ public void testScanBucketOptionViaDirectTableScan() throws Exception {
+ FileStoreTable table = createTable("4", true);
+ writeRows(table, 1, 2, 3, 4, 5, 6, 7, 8);
+
+
assertThat(extractBuckets(table.newScan().plan().splits()).size()).isGreaterThan(1);
+
+ Map<String, String> options = new HashMap<>(table.options());
+ options.put(CoreOptions.SCAN_BUCKET.key(), "0");
+ FileStoreTable tableWithScanBucket = table.copy(options);
+
assertThat(extractBuckets(tableWithScanBucket.newScan().plan().splits()))
+ .containsExactly(0);
+ }
+
+ @Test
+ public void testScanBucketOptionViaTableCopy() throws Exception {
+ FileStoreTable table = createTable("4", true);
+ writeRows(table, 1, 2, 3, 4, 5, 6, 7, 8);
+
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.SCAN_BUCKET.key(), "0");
+ FileStoreTable tableWithScanBucket = table.copy(options);
+
+
assertThat(extractBuckets(table.newScan().plan().splits()).size()).isGreaterThan(1);
+
assertThat(extractBuckets(tableWithScanBucket.newScan().plan().splits()))
+ .containsExactly(0);
+ }
+
+ @Test
+ public void testScanBucketOptionViaReadBuilder() throws Exception {
+ FileStoreTable table = createTableWithScanBucket("4", true, "0");
+ writeRows(table, 1, 2, 3, 4, 5, 6, 7, 8);
+
+
assertThat(extractBuckets(table.newReadBuilder().newScan().plan().splits()))
+ .containsExactly(0);
+ }
+
+ @Test
+ public void
testScanBucketOptionRejectsDirectTableScanForDynamicBucketTable() throws
Exception {
+ FileStoreTable table = createTableWithScanBucket("-1", true, "0");
+ assertThatThrownBy(() -> table.newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("fixed-bucket tables");
+ }
+
+ private static List<Integer> extractBuckets(List<Split> splits) {
+ return splits.stream()
+ .map(split -> ((DataSplit) split).bucket())
+ .distinct()
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private void writeRows(FileStoreTable table, int... ids) throws Exception {
+ StreamTableWrite write = table.newWrite("test");
+ StreamTableCommit commit = table.newCommit("test");
+ for (int id : ids) {
+ write.write(GenericRow.of(id, id));
+ }
+ commit.commit(0, write.prepareCommit(true, 0));
+ write.close();
+ commit.close();
+ }
+
+ private FileStoreTable createTableWithScanBucket(
+ String bucket, boolean withPrimaryKey, String scanBucket) throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), bucket);
+ options.put(CoreOptions.SCAN_BUCKET.key(), scanBucket);
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder().column("id", DataTypes.INT()).column("v",
DataTypes.INT());
+ if (withPrimaryKey) {
+ schemaBuilder.primaryKey("id");
+ }
+ Schema schema = schemaBuilder.options(options).build();
+ return createTable(schema);
+ }
+
+ private FileStoreTable createBucketUnawareTableWithScanBucket(String
scanBucket)
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+ options.put(CoreOptions.SCAN_BUCKET.key(), scanBucket);
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.INT())
+ .options(options)
+ .build();
+ return createTable(schema);
+ }
+
+ private FileStoreTable createAppendOnlyFixedBucketTable(String bucket)
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), bucket);
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.INT())
+ .options(options)
+ .build();
+ return createTable(schema);
+ }
+
+ private FileStoreTable createAppendOnlyTableWithScanBucket(String bucket,
String scanBucket)
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), bucket);
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+ options.put(CoreOptions.SCAN_BUCKET.key(), scanBucket);
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("v", DataTypes.INT())
+ .options(options)
+ .build();
+ return createTable(schema);
+ }
+
+ private FileStoreTable createTable(String bucket, boolean withPrimaryKey)
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), bucket);
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder().column("id", DataTypes.INT()).column("v",
DataTypes.INT());
+ if (withPrimaryKey) {
+ schemaBuilder.primaryKey("id");
+ }
+ Schema schema = schemaBuilder.options(options).build();
+ return createTable(schema);
+ }
+
+ private FileStoreTable createTable(Schema schema) throws Exception {
+ Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new
Path(tempDir.toString()));
+ catalog.createDatabase("default", true);
+ Identifier identifier = Identifier.create("default", "test_bucket");
+ catalog.createTable(identifier, schema, false);
+ return (FileStoreTable) catalog.getTable(identifier);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ScanBucketITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ScanBucketITCase.java
new file mode 100644
index 0000000000..4e4ae6a525
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ScanBucketITCase.java
@@ -0,0 +1,174 @@
+/*
+ * 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.paimon.flink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static java.util.Collections.singletonList;
+import static
org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for {@link CoreOptions#SCAN_BUCKET}. */
+public class ScanBucketITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ return singletonList(
+ "CREATE TABLE IF NOT EXISTS T (id INT, v INT, PRIMARY KEY (id)
NOT ENFORCED) "
+ + "WITH ('bucket' = '4')");
+ }
+
+ @Nullable
+ @Override
+ protected Boolean sqlSyncMode() {
+ return true;
+ }
+
+ @Test
+ public void testScanBucketFilter() throws Exception {
+ FileStoreTable table = paimonTable("T");
+ writeRows(table, 1, 10, 2, 20, 3, 30, 4, 40, 5, 50, 6, 60, 7, 70, 8,
80);
+
+ int targetBucket = 0;
+ for (int bucket = 0; bucket < 4; bucket++) {
+ List<ManifestEntry> files =
table.store().newScan().withBucket(bucket).plan().files();
+ if (!files.isEmpty()) {
+ targetBucket = bucket;
+ break;
+ }
+ }
+
+ List<Row> expected = readRowsFromBucket(table, targetBucket);
+
+ List<Row> actual =
+ batchSql(
+ String.format(
+ "SELECT * FROM T /*+ OPTIONS('scan.bucket' =
'%s') */",
+ targetBucket));
+
+ assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
+
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT COUNT(*) FROM T /*+
OPTIONS('scan.bucket' = '%s') */",
+ targetBucket)))
+ .containsExactly(Row.of((long) expected.size()));
+ }
+
+ @Test
+ public void testScanBucketRejectsDynamicBucketTable() {
+ sql(
+ "CREATE TABLE dynamic_t (id INT, v INT, PRIMARY KEY (id) NOT
ENFORCED) "
+ + "WITH ('bucket' = '-1')");
+
+ assertThatThrownBy(
+ () ->
+ batchSql(
+ "SELECT * FROM dynamic_t /*+
OPTIONS('scan.bucket' = '0') */"))
+ .satisfies(
+ anyCauseMatches(
+ IllegalArgumentException.class,
+ "Bucket scan is only supported for
fixed-bucket tables"));
+ }
+
+ @Test
+ public void testScanBucketRejectsBucketUnawareTable() {
+ sql("CREATE TABLE append_t (id INT, v INT) WITH ('bucket' = '-1')");
+
+ assertThatThrownBy(
+ () ->
+ batchSql(
+ "SELECT * FROM append_t /*+
OPTIONS('scan.bucket' = '0') */"))
+ .satisfies(
+ anyCauseMatches(
+ IllegalArgumentException.class,
+ "Bucket scan is only supported for primary key
tables"));
+ }
+
+ @Test
+ public void testScanBucketRejectsPostponeBucketTable() {
+ sql(
+ "CREATE TABLE postpone_t (id INT, v INT, PRIMARY KEY (id) NOT
ENFORCED) "
+ + "WITH ('bucket' = '-2')");
+
+ assertThatThrownBy(
+ () ->
+ batchSql(
+ "SELECT * FROM postpone_t /*+
OPTIONS('scan.bucket' = '0') */"))
+ .satisfies(
+ anyCauseMatches(
+ IllegalArgumentException.class,
+ "Bucket scan is only supported for
fixed-bucket tables"));
+ }
+
+ private void writeRows(FileStoreTable table, int... idAndValues) throws
Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ BatchTableWrite write = writeBuilder.newWrite();
+ for (int i = 0; i < idAndValues.length; i += 2) {
+ write.write(GenericRow.of(idAndValues[i], idAndValues[i + 1]));
+ }
+ BatchTableCommit commit = writeBuilder.newCommit();
+ commit.commit(write.prepareCommit());
+ write.close();
+ commit.close();
+ }
+
+ private List<Row> readRowsFromBucket(FileStoreTable table, int bucket)
throws Exception {
+ List<ManifestEntry> files =
table.store().newScan().withBucket(bucket).plan().files();
+ List<Row> rows = new ArrayList<>();
+ for (ManifestEntry file : files) {
+ DataSplit split =
+ DataSplit.builder()
+ .withPartition(file.partition())
+ .withBucket(file.bucket())
+
.withDataFiles(Collections.singletonList(file.file()))
+ .withBucketPath("not used")
+ .build();
+ RecordReader<InternalRow> reader =
table.newReadBuilder().newRead().createReader(split);
+ RecordReaderIterator<InternalRow> iterator = new
RecordReaderIterator<>(reader);
+ while (iterator.hasNext()) {
+ InternalRow row = iterator.next();
+ rows.add(Row.of(row.getInt(0), row.getInt(1)));
+ }
+ iterator.close();
+ }
+ return rows;
+ }
+}
