You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/02/01 03:33:37 UTC
(incubator-paimon) branch master updated: [core] Adjust default value of bucket to -1 (#2829)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4b667aab3 [core] Adjust default value of bucket to -1 (#2829)
4b667aab3 is described below
commit 4b667aab3a69f19f1264f276a96b4d4730f025b4
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Feb 1 11:33:33 2024 +0800
[core] Adjust default value of bucket to -1 (#2829)
---
.../shortcodes/generated/core_configuration.html | 2 +-
.../main/java/org/apache/paimon/CoreOptions.java | 2 +-
.../org/apache/paimon/schema/SchemaSerializer.java | 13 +++++++++++
.../java/org/apache/paimon/schema/TableSchema.java | 23 +++++++++++++++---
.../table/sink/DynamicBucketRowKeyExtractor.java | 4 +++-
.../paimon/index/HashIndexMaintainerTest.java | 4 +++-
.../schema/TableSchemaSerializationTest.java | 27 ++++++++++++++++++----
.../org/apache/paimon/schema/TableSchemaTest.java | 1 +
.../org/apache/paimon/stats/StatsTableTest.java | 2 ++
.../apache/paimon/table/IncrementalTableTest.java | 3 +++
.../table/IncrementalTimeStampTableTest.java | 3 +++
.../paimon/table/PrimaryKeyFileStoreTableTest.java | 10 ++++----
.../paimon/table/WritePreemptMemoryTest.java | 14 ++++++-----
.../apache/paimon/table/sink/TableCommitTest.java | 7 +++---
.../table/source/snapshot/ScannerTestBase.java | 4 ++++
.../paimon/table/system/AuditLogTableTest.java | 1 +
.../paimon/table/system/ManifestsTableTest.java | 1 +
.../paimon/table/system/PartitionsTableTest.java | 1 +
.../cdc/mysql/MySqlCdcTypeMappingITCase.java | 5 ++++
.../apache/paimon/flink/kafka/LogSystemITCase.java | 12 +++++++++-
.../flink/kafka/StreamingWarehouseITCase.java | 1 +
.../cdc/CdcRecordKeyAndBucketExtractorTest.java | 3 ++-
.../cdc/CdcRecordStoreMultiWriteOperatorTest.java | 2 ++
.../sink/cdc/CdcRecordStoreWriteOperatorTest.java | 2 ++
.../flink/sink/GlobalFullCompactionSinkWrite.java | 7 ++++++
.../apache/paimon/flink/AppendOnlyTableITCase.java | 6 ++---
.../apache/paimon/flink/BatchFileStoreITCase.java | 4 ++--
.../apache/paimon/flink/CatalogTableITCase.java | 9 +++++++-
.../paimon/flink/ContinuousFileStoreITCase.java | 7 +++---
.../paimon/flink/FileSystemCatalogITCase.java | 6 ++---
.../apache/paimon/flink/ReadWriteTableITCase.java | 1 +
.../paimon/flink/RemoteLookupJoinITCase.java | 3 ++-
.../apache/paimon/flink/SchemaChangeITCase.java | 5 ++--
.../paimon/flink/action/ActionITCaseBase.java | 7 +++++-
.../flink/action/CompactDatabaseActionITCase.java | 3 +++
.../flink/sink/CommitterOperatorTestBase.java | 1 +
.../paimon/flink/sink/CompactorSinkITCase.java | 4 ++--
.../apache/paimon/flink/sink/FlinkSinkTest.java | 12 +++++-----
.../flink/sink/PrimaryKeyWriterOperatorTest.java | 1 +
.../flink/sink/RowDataChannelComputerTest.java | 4 ++--
.../paimon/flink/sink/StoreMultiCommitterTest.java | 3 ++-
.../paimon/flink/source/CompactorSourceITCase.java | 2 +-
.../paimon/flink/source/DataTableSourceTest.java | 2 ++
.../MultiTablesCompactorSourceBuilderITCase.java | 3 +++
.../flink/source/operator/OperatorSourceTest.java | 4 ++--
.../statistics/PrimaryKeyTableStatisticsTest.java | 12 ++++++----
.../paimon/flink/util/ReadWriteTableTestUtil.java | 6 ++++-
.../org/apache/paimon/hive/HiveWriteITCase.java | 1 +
.../paimon/hive/mapred/PaimonRecordReaderTest.java | 1 +
.../apache/paimon/service/KvQueryTableTest.java | 8 +++++++
.../apache/paimon/spark/SimpleTableTestHelper.java | 1 +
.../org/apache/paimon/spark/SparkReadTestBase.java | 2 +-
.../apache/paimon/spark/SparkTimeTravelITCase.java | 2 +-
.../paimon/spark/sql/DeleteFromTableTest.scala | 18 ++++++++-------
.../paimon/spark/sql/LookupCompactionTest.scala | 2 +-
55 files changed, 221 insertions(+), 73 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index f9d167de4..29d6157e6 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -34,7 +34,7 @@ under the License.
</tr>
<tr>
<td><h5>bucket</h5></td>
- <td style="word-wrap: break-word;">1</td>
+ <td style="word-wrap: break-word;">-1</td>
<td>Integer</td>
<td>Bucket number for file store.</td>
</tr>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 6e7730928..ba238ac99 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -71,7 +71,7 @@ public class CoreOptions implements Serializable {
public static final ConfigOption<Integer> BUCKET =
key("bucket")
.intType()
- .defaultValue(1)
+ .defaultValue(-1)
.withDescription("Bucket number for file store.");
@Immutable
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaSerializer.java
index ca7fc540b..dd77497fe 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaSerializer.java
@@ -33,6 +33,9 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import static org.apache.paimon.CoreOptions.BUCKET;
+import static org.apache.paimon.schema.TableSchema.PAIMON_07_VERSION;
+
/** A {@link JsonSerializer} for {@link TableSchema}. */
public class SchemaSerializer
implements JsonSerializer<TableSchema>, JsonDeserializer<TableSchema> {
@@ -43,6 +46,8 @@ public class SchemaSerializer
public void serialize(TableSchema tableSchema, JsonGenerator generator) throws IOException {
generator.writeStartObject();
+ generator.writeNumberField("version", tableSchema.version());
+
generator.writeNumberField("id", tableSchema.id());
generator.writeArrayFieldStart("fields");
@@ -82,6 +87,9 @@ public class SchemaSerializer
@Override
public TableSchema deserialize(JsonNode node) {
+ JsonNode versionNode = node.get("version");
+ int version = versionNode == null ? PAIMON_07_VERSION : versionNode.asInt();
+
int id = node.get("id").asInt();
Iterator<JsonNode> fieldJsons = node.get("fields").elements();
@@ -111,6 +119,10 @@ public class SchemaSerializer
String key = optionsKeys.next();
options.put(key, optionsJson.get(key).asText());
}
+ if (version == PAIMON_07_VERSION && !options.containsKey(BUCKET.key())) {
+ // the default value of BUCKET in old version is 1
+ options.put(BUCKET.key(), "1");
+ }
JsonNode commentNode = node.get("comment");
String comment = null;
@@ -121,6 +133,7 @@ public class SchemaSerializer
long timeMillis = node.get("timeMillis") == null ? 0 : node.get("timeMillis").asLong();
return new TableSchema(
+ version,
id,
fields,
highestFieldId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java b/paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java
index cd8326d54..6fe65aca1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java
@@ -45,6 +45,12 @@ public class TableSchema implements Serializable {
private static final long serialVersionUID = 1L;
+ public static final int PAIMON_07_VERSION = 1;
+ public static final int CURRENT_VERSION = 2;
+
+ // version of schema for paimon
+ private final int version;
+
private final long id;
private final List<DataField> fields;
@@ -71,6 +77,7 @@ public class TableSchema implements Serializable {
Map<String, String> options,
@Nullable String comment) {
this(
+ CURRENT_VERSION,
id,
fields,
highestFieldId,
@@ -82,6 +89,7 @@ public class TableSchema implements Serializable {
}
public TableSchema(
+ int version,
long id,
List<DataField> fields,
int highestFieldId,
@@ -90,6 +98,7 @@ public class TableSchema implements Serializable {
Map<String, String> options,
@Nullable String comment,
long timeMillis) {
+ this.version = version;
this.id = id;
this.fields = fields;
this.highestFieldId = highestFieldId;
@@ -106,6 +115,10 @@ public class TableSchema implements Serializable {
originalBucketKeys();
}
+ public int version() {
+ return version;
+ }
+
public long id() {
return id;
}
@@ -261,6 +274,7 @@ public class TableSchema implements Serializable {
public TableSchema copy(Map<String, String> newOptions) {
return new TableSchema(
+ version,
id,
fields,
highestFieldId,
@@ -285,16 +299,19 @@ public class TableSchema implements Serializable {
return false;
}
TableSchema tableSchema = (TableSchema) o;
- return Objects.equals(fields, tableSchema.fields)
+ return version == tableSchema.version
+ && Objects.equals(fields, tableSchema.fields)
&& Objects.equals(partitionKeys, tableSchema.partitionKeys)
&& Objects.equals(primaryKeys, tableSchema.primaryKeys)
&& Objects.equals(options, tableSchema.options)
- && Objects.equals(comment, tableSchema.comment);
+ && Objects.equals(comment, tableSchema.comment)
+ && timeMillis == tableSchema.timeMillis;
}
@Override
public int hashCode() {
- return Objects.hash(fields, partitionKeys, primaryKeys, options, comment);
+ return Objects.hash(
+ version, fields, partitionKeys, primaryKeys, options, comment, timeMillis);
}
public static List<DataField> newFields(RowType rowType) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/DynamicBucketRowKeyExtractor.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/DynamicBucketRowKeyExtractor.java
index 5f4734bf5..949087837 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/DynamicBucketRowKeyExtractor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/DynamicBucketRowKeyExtractor.java
@@ -41,6 +41,8 @@ public class DynamicBucketRowKeyExtractor extends RowKeyExtractor {
@Override
public int bucket() {
- throw new IllegalArgumentException("Can't extract bucket from row in dynamic bucket mode.");
+ throw new IllegalArgumentException(
+ "Can't extract bucket from row in dynamic bucket mode, "
+ + "you should use 'TableWrite.write(InternalRow row, int bucket)' method.");
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/HashIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/HashIndexMaintainerTest.java
index 043e3c727..c4b28d77e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/HashIndexMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/HashIndexMaintainerTest.java
@@ -90,7 +90,9 @@ public class HashIndexMaintainerTest extends PrimaryKeyTableTestBase {
@Test
public void testAssignBucket() throws Exception {
assertThatThrownBy(() -> write.write(GenericRow.of(1, 1, 1)))
- .hasMessageContaining("Can't extract bucket from row in dynamic bucket mode.");
+ .hasMessageContaining(
+ "Can't extract bucket from row in dynamic bucket mode, "
+ + "you should use 'TableWrite.write(InternalRow row, int bucket)' method.");
// commit two partitions
write(write, createRow(1, 1, 1, 1));
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaSerializationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaSerializationTest.java
index 12bd95e6a..4657f0f89 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaSerializationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaSerializationTest.java
@@ -33,6 +33,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.paimon.schema.TableSchema.PAIMON_07_VERSION;
import static org.apache.paimon.schema.TableSchemaTest.newRowType;
import static org.assertj.core.api.Assertions.assertThat;
@@ -57,11 +58,29 @@ public class TableSchemaSerializationTest {
options.put("option-1", "value-1");
options.put("option-2", "value-2");
- TableSchema tableSchema =
- new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, "my_comment");
- String serialized = JsonSerdeUtil.toJson(tableSchema);
+ assertSerDeser(
+ new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, "my_comment"),
+ Collections.emptyMap());
+ assertSerDeser(
+ new TableSchema(
+ PAIMON_07_VERSION,
+ 1,
+ fields,
+ 10,
+ partitionKeys,
+ primaryKeys,
+ options,
+ "my_comment",
+ System.currentTimeMillis()),
+ Collections.singletonMap("bucket", "1"));
+ }
+ private void assertSerDeser(TableSchema tableSchema, Map<String, String> additionalOptions) {
+ String serialized = JsonSerdeUtil.toJson(tableSchema);
TableSchema deserialized = JsonSerdeUtil.fromJson(serialized, TableSchema.class);
- assertThat(deserialized).isEqualTo(tableSchema);
+ Map<String, String> newOptions = new HashMap<>(tableSchema.options());
+ newOptions.putAll(additionalOptions);
+ assertThat(deserialized.version()).isEqualTo(tableSchema.version());
+ assertThat(deserialized).isEqualTo(tableSchema.copy(newOptions));
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaTest.java
index e0fc6b6d9..d735ea819 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/TableSchemaTest.java
@@ -54,6 +54,7 @@ public class TableSchemaTest {
new TableSchema(1, fields, 10, partitionKeys, primaryKeys, options, "");
assertThat(schema.crossPartitionUpdate()).isTrue();
+ options.put(BUCKET.key(), "1");
assertThatThrownBy(() -> validateTableSchema(schema))
.hasMessageContaining("You should use dynamic bucket");
diff --git a/paimon-core/src/test/java/org/apache/paimon/stats/StatsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/stats/StatsTableTest.java
index e63d481d9..b51537972 100644
--- a/paimon-core/src/test/java/org/apache/paimon/stats/StatsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/stats/StatsTableTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.stats;
import org.apache.paimon.AbstractFileStore;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.io.DataFileMeta;
@@ -45,6 +46,7 @@ public class StatsTableTest extends TableTestBase {
Identifier identifier = identifier("T");
Options options = new Options();
options.set(METADATA_STATS_MODE, "NONE");
+ options.set(CoreOptions.BUCKET, 1);
Schema schema =
Schema.newBuilder()
.column("pt", DataTypes.INT())
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
index a7602bb72..f4dac9d11 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
@@ -50,6 +50,7 @@ public class IncrementalTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
@@ -161,6 +162,7 @@ public class IncrementalTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
@@ -211,6 +213,7 @@ public class IncrementalTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTimeStampTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTimeStampTableTest.java
index 1e64c7e0f..13c9f8783 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTimeStampTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTimeStampTableTest.java
@@ -51,6 +51,7 @@ public class IncrementalTimeStampTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
@@ -139,6 +140,7 @@ public class IncrementalTimeStampTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
@@ -176,6 +178,7 @@ public class IncrementalTimeStampTableTest extends TableTestBase {
.column("col1", DataTypes.INT())
.partitionKeys("pt")
.primaryKey("pk", "pt")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
index c3dce3a8a..92172ae4f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
@@ -1251,6 +1251,7 @@ public class PrimaryKeyFileStoreTableTest extends FileStoreTableTestBase {
protected FileStoreTable overwriteTestFileStoreTable() throws Exception {
Options conf = new Options();
conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(BUCKET, 1);
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(LocalFileIO.create(), tablePath),
@@ -1265,9 +1266,10 @@ public class PrimaryKeyFileStoreTableTest extends FileStoreTableTestBase {
private FileStoreTable createFileStoreTable(Consumer<Options> configure, RowType rowType)
throws Exception {
- Options conf = new Options();
- conf.set(CoreOptions.PATH, tablePath.toString());
- configure.accept(conf);
+ Options options = new Options();
+ options.set(CoreOptions.PATH, tablePath.toString());
+ options.set(BUCKET, 1);
+ configure.accept(options);
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(LocalFileIO.create(), tablePath),
@@ -1275,7 +1277,7 @@ public class PrimaryKeyFileStoreTableTest extends FileStoreTableTestBase {
rowType.getFields(),
Collections.singletonList("pt"),
Arrays.asList("pt", "a"),
- conf.toMap(),
+ options.toMap(),
""));
return new PrimaryKeyFileStoreTable(FileIOFinder.find(tablePath), tablePath, tableSchema);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/WritePreemptMemoryTest.java b/paimon-core/src/test/java/org/apache/paimon/table/WritePreemptMemoryTest.java
index adb0c6b8b..548dc7205 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/WritePreemptMemoryTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/WritePreemptMemoryTest.java
@@ -89,14 +89,15 @@ public class WritePreemptMemoryTest extends FileStoreTableTestBase {
@Override
protected FileStoreTable createFileStoreTable(Consumer<Options> configure) throws Exception {
- Options conf = new Options();
- conf.set(CoreOptions.PATH, tablePath.toString());
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.PATH, tablePath.toString());
// Run with minimal memory to ensure a more intense preempt
// Currently a writer needs at least one page
int pages = 10;
- conf.set(CoreOptions.WRITE_BUFFER_SIZE, new MemorySize(pages * 1024));
- conf.set(CoreOptions.PAGE_SIZE, new MemorySize(1024));
- configure.accept(conf);
+ options.set(CoreOptions.WRITE_BUFFER_SIZE, new MemorySize(pages * 1024));
+ options.set(CoreOptions.PAGE_SIZE, new MemorySize(1024));
+ configure.accept(options);
TableSchema schema =
SchemaUtils.forceCommit(
new SchemaManager(LocalFileIO.create(), tablePath),
@@ -104,7 +105,7 @@ public class WritePreemptMemoryTest extends FileStoreTableTestBase {
ROW_TYPE.getFields(),
Collections.singletonList("pt"),
Arrays.asList("pt", "a"),
- conf.toMap(),
+ options.toMap(),
""));
return new PrimaryKeyFileStoreTable(FileIOFinder.find(tablePath), tablePath, schema);
}
@@ -113,6 +114,7 @@ public class WritePreemptMemoryTest extends FileStoreTableTestBase {
protected FileStoreTable overwriteTestFileStoreTable() throws Exception {
Options conf = new Options();
conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.set(CoreOptions.BUCKET, 1);
// Run with minimal memory to ensure a more intense preempt
// Currently a writer needs at least one page
int pages = 10;
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java
index 362b40301..bed06293a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java
@@ -194,8 +194,9 @@ public class TableCommitTest {
new DataType[] {DataTypes.INT(), DataTypes.BIGINT()},
new String[] {"k", "v"});
- Options conf = new Options();
- conf.set(CoreOptions.PATH, path);
+ Options options = new Options();
+ options.set(CoreOptions.PATH, path);
+ options.set(CoreOptions.BUCKET, 1);
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(LocalFileIO.create(), new Path(path)),
@@ -203,7 +204,7 @@ public class TableCommitTest {
rowType.getFields(),
Collections.emptyList(),
Collections.singletonList("k"),
- conf.toMap(),
+ options.toMap(),
""));
FileStoreTable table =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ScannerTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ScannerTestBase.java
index 7fea1b42d..3e37ee0e4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ScannerTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/ScannerTestBase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source.snapshot;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.GenericRow;
@@ -152,6 +153,9 @@ public abstract class ScannerTestBase {
if (withPrimaryKeys) {
primaryKeys = Arrays.asList("pt", "a");
}
+ if (!conf.contains(CoreOptions.BUCKET)) {
+ conf.set(CoreOptions.BUCKET, 1);
+ }
TableSchema tableSchema =
schemaManager.createTable(
new Schema(
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java
index fc8874a0b..77d2cafe7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/AuditLogTableTest.java
@@ -64,6 +64,7 @@ public class AuditLogTableTest extends TableTestBase {
.partitionKeys("pt")
.primaryKey("pk", "pt")
.option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
+ .option("bucket", "1")
.build();
TableSchema tableSchema =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
index edb35fbb0..fa659430c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/ManifestsTableTest.java
@@ -66,6 +66,7 @@ public class ManifestsTableTest extends TableTestBase {
.partitionKeys("pt")
.primaryKey("pk", "pt")
.option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
+ .option("bucket", "1")
.build();
catalog.createTable(identifier, schema, true);
table = catalog.getTable(identifier);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
index 27591b9ca..4acfae7ed 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
@@ -67,6 +67,7 @@ public class PartitionsTableTest extends TableTestBase {
.partitionKeys("pt")
.primaryKey("pk", "pt")
.option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
+ .option("bucket", "1")
.build();
TableSchema tableSchema =
SchemaUtils.forceCommit(new SchemaManager(fileIO, tablePath), schema);
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
index 8a497d442..38a99bc2a 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlCdcTypeMappingITCase.java
@@ -343,6 +343,7 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
.excludingTables("all_types_table")
.withMode(COMBINED.configString())
.withTypeMappingModes(TO_STRING.configString())
+ .withTableConfig(getBasicTableConfig())
.build();
runActionWithDefaultEnv(action);
@@ -419,6 +420,7 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
syncDatabaseActionBuilder(mySqlConfig)
.withMode(COMBINED.configString())
.withTypeMappingModes(TO_NULLABLE.configString())
+ .withTableConfig(getBasicTableConfig())
.build();
runActionWithDefaultEnv(action);
@@ -477,6 +479,7 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
syncDatabaseActionBuilder(mySqlConfig)
.withMode(COMBINED.configString())
.withTypeMappingModes(CHAR_TO_STRING.configString())
+ .withTableConfig(getBasicTableConfig())
.build();
runActionWithDefaultEnv(action);
@@ -543,6 +546,7 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
syncDatabaseActionBuilder(mySqlConfig)
.withMode(COMBINED.configString())
.withTypeMappingModes(LONGTEXT_TO_BYTES.configString())
+ .withTableConfig(getBasicTableConfig())
.build();
runActionWithDefaultEnv(action);
@@ -612,6 +616,7 @@ public class MySqlCdcTypeMappingITCase extends MySqlActionITCaseBase {
syncDatabaseActionBuilder(mySqlConfig)
.withMode(COMBINED.configString())
.withTypeMappingModes(BIGINT_UNSIGNED_TO_BIGINT.configString())
+ .withTableConfig(getBasicTableConfig())
.build();
runActionWithDefaultEnv(action);
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/LogSystemITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/LogSystemITCase.java
index 8f2c0a820..b1dadced9 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/LogSystemITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/LogSystemITCase.java
@@ -65,6 +65,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE T (i INT, j INT) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'log.consistency'='eventual', "
+ "'kafka.bootstrap.servers'='%s', "
+ "'kafka.topic'='T')",
@@ -95,6 +96,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
+ " 'merge-engine' = 'aggregation',\n"
+ " 'changelog-producer' = 'full-compaction',\n"
+ " 'log.system' = 'kafka',\n"
+ + " 'bucket'='1',\n"
+ " 'streaming-read-mode'='file',\n"
+ " 'fields.cnt.aggregate-function' = 'sum',\n"
+ " 'kafka.bootstrap.servers' = '%s',\n"
@@ -137,6 +139,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
+ " 'changelog-producer' = 'full-compaction',\n"
+ " 'log.consistency' = 'eventual',\n"
+ " 'log.system' = 'kafka',\n"
+ + " 'bucket'='1',\n"
+ " 'streaming-read-mode'='log',\n"
+ " 'kafka.bootstrap.servers' = '%s',\n"
+ " 'kafka.topic' = 'test-single-sink',\n"
@@ -175,7 +178,8 @@ public class LogSystemITCase extends KafkaTableTestBase {
+ " PRIMARY KEY (word) NOT ENFORCED\n"
+ ")\n"
+ "WITH (\n"
- + " 'merge-engine' = 'aggregation',\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'bucket'='1',\n"
+ " 'changelog-producer' = 'full-compaction',\n"
+ " 'streaming-read-mode'='log'\n"
+ ");");
@@ -215,6 +219,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE T (i INT, j INT) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'log.system.partitions'='2', "
+ "'kafka.bootstrap.servers'='%s', "
+ "'kafka.topic'='Tt')",
@@ -238,6 +243,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE T1 (i INT, j INT) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'log.system.partitions'='2', "
+ "'kafka.bootstrap.servers'='%s')",
getBootstrapServers()));
@@ -255,6 +261,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE T (i INT, j INT) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'log.system.partitions'='2', "
+ "'kafka.bootstrap.servers'='%s', "
+ "'kafka.topic'='T1')",
@@ -275,6 +282,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE NOT_EXIST.T (i INT, j INT) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'log.system.partitions'='2', "
+ "'kafka.bootstrap.servers'='%s', "
+ "'kafka.topic'='T1')",
@@ -328,6 +336,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
String.format(
"CREATE TABLE T (a STRING, b STRING, c STRING) WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'kafka.bootstrap.servers'='%s',"
+ "'kafka.topic'='%s'"
+ ")",
@@ -359,6 +368,7 @@ public class LogSystemITCase extends KafkaTableTestBase {
+ "d AS CAST(c as INT) + 1"
+ ") WITH ("
+ "'log.system'='kafka', "
+ + "'bucket'='1', "
+ "'kafka.bootstrap.servers'='%s',"
+ "'kafka.topic'='%s'"
+ ")",
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/StreamingWarehouseITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/StreamingWarehouseITCase.java
index 9c21f491c..4d308f54c 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/StreamingWarehouseITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/kafka/StreamingWarehouseITCase.java
@@ -82,6 +82,7 @@ public class StreamingWarehouseITCase extends KafkaTableTestBase {
+ " )\n"
+ "PARTITIONED BY (dt)\n"
+ "WITH (\n"
+ + " 'bucket' = '1',\n"
+ " 'log.system' = 'kafka',\n"
+ " 'kafka.bootstrap.servers' = '%s',\n"
+ " 'kafka.topic' = 'cleaned_trade_order');",
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordKeyAndBucketExtractorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordKeyAndBucketExtractorTest.java
index 4ec0215ed..a9dc3907d 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordKeyAndBucketExtractorTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordKeyAndBucketExtractorTest.java
@@ -36,6 +36,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -145,7 +146,7 @@ public class CdcRecordKeyAndBucketExtractorTest {
ROW_TYPE.getFields(),
Arrays.asList("pt1", "pt2"),
Arrays.asList("pt1", "pt2", "k1", "k2"),
- new HashMap<>(),
+ Collections.singletonMap("bucket", "1"),
""));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
index 23ca641c3..ff4132103 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.sink.cdc;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
@@ -101,6 +102,7 @@ public class CdcRecordStoreMultiWriteOperatorTest {
catalog.createDatabase(databaseName, true);
Options conf = new Options();
conf.set(CdcRecordStoreWriteOperator.RETRY_SLEEP_TIME, Duration.ofMillis(10));
+ conf.set(CoreOptions.BUCKET, 1);
RowType rowType1 =
RowType.of(
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
index 87538e047..8d03466f6 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreWriteOperatorTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.sink.cdc;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.CommittableTypeInfo;
@@ -355,6 +356,7 @@ public class CdcRecordStoreWriteOperatorTest {
RowType rowType, List<String> partitions, List<String> primaryKeys) throws Exception {
Options conf = new Options();
conf.set(CdcRecordStoreWriteOperator.RETRY_SLEEP_TIME, Duration.ofMillis(10));
+ conf.set(CoreOptions.BUCKET, 1);
TableSchema tableSchema =
SchemaUtils.forceCommit(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/GlobalFullCompactionSinkWrite.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/GlobalFullCompactionSinkWrite.java
index 317d80a38..3fb6fe6ae 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/GlobalFullCompactionSinkWrite.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/GlobalFullCompactionSinkWrite.java
@@ -114,6 +114,13 @@ public class GlobalFullCompactionSinkWrite extends StoreSinkWriteImpl {
return sinkRecord;
}
+ @Override
+ public SinkRecord write(InternalRow rowData, int bucket) throws Exception {
+ SinkRecord sinkRecord = super.write(rowData, bucket);
+ touchBucket(sinkRecord.partition(), bucket);
+ return sinkRecord;
+ }
+
@Override
public void compact(BinaryRow partition, int bucket, boolean fullCompaction) throws Exception {
super.compact(partition, bucket, fullCompaction);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java
index 0dc141ee6..11a6b941f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java
@@ -262,9 +262,9 @@ public class AppendOnlyTableITCase extends CatalogITCaseBase {
@Override
protected List<String> ddl() {
return Arrays.asList(
- "CREATE TABLE IF NOT EXISTS append_table (id INT, data STRING)",
- "CREATE TABLE IF NOT EXISTS part_table (id INT, data STRING, dt STRING) PARTITIONED BY (dt)",
- "CREATE TABLE IF NOT EXISTS complex_table (id INT, data MAP<INT, INT>)");
+ "CREATE TABLE IF NOT EXISTS append_table (id INT, data STRING) WITH ('bucket' = '1')",
+ "CREATE TABLE IF NOT EXISTS part_table (id INT, data STRING, dt STRING) PARTITIONED BY (dt) WITH ('bucket' = '1')",
+ "CREATE TABLE IF NOT EXISTS complex_table (id INT, data MAP<INT, INT>) WITH ('bucket' = '1')");
}
private void testRejectChanges(RowKind kind) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 7e2432d08..9829e1040 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -375,7 +375,7 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
public void testIgnoreDelete() throws Exception {
sql(
"CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED, v STRING) "
- + "WITH ('deduplicate.ignore-delete' = 'true')");
+ + "WITH ('deduplicate.ignore-delete' = 'true', 'bucket' = '1')");
BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT * FROM ignore_delete");
sql("INSERT INTO ignore_delete VALUES (1, 'A'), (2, 'B')");
@@ -392,7 +392,7 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
public void testDeleteWithPkLookup() throws Exception {
sql(
"CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED, v STRING) "
- + "WITH ('changelog-producer' = 'lookup')");
+ + "WITH ('changelog-producer' = 'lookup', 'bucket' = '1')");
BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT * FROM ignore_delete");
sql("INSERT INTO ignore_delete VALUES (1, 'A'), (2, 'B')");
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 150caf7b0..60407710b 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -781,8 +781,15 @@ public class CatalogTableITCase extends CatalogITCaseBase {
@Test
public void testReadOptimizedTable() {
- sql("CREATE TABLE T (k INT, v INT, PRIMARY KEY (k) NOT ENFORCED)");
+ sql("CREATE TABLE T (k INT, v INT, PRIMARY KEY (k) NOT ENFORCED) WITH ('bucket' = '1')");
+ innerTestReadOptimizedTable();
+ sql("DROP TABLE T");
+ sql("CREATE TABLE T (k INT, v INT, PRIMARY KEY (k) NOT ENFORCED) WITH ('bucket' = '-1')");
+ innerTestReadOptimizedTable();
+ }
+
+ private void innerTestReadOptimizedTable() {
// full compaction will always be performed at the end of batch jobs, as long as
// full-compaction.delta-commits is set, regardless of its value
sql(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
index 811abc673..8ba3221d3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
@@ -49,8 +49,9 @@ public class ContinuousFileStoreITCase extends CatalogITCaseBase {
@Override
protected List<String> ddl() {
return Arrays.asList(
- "CREATE TABLE IF NOT EXISTS T1 (a STRING, b STRING, c STRING)",
- "CREATE TABLE IF NOT EXISTS T2 (a STRING, b STRING, c STRING, PRIMARY KEY (a) NOT ENFORCED) WITH('changelog-producer'='input')");
+ "CREATE TABLE IF NOT EXISTS T1 (a STRING, b STRING, c STRING) WITH ('bucket' = '1')",
+ "CREATE TABLE IF NOT EXISTS T2 (a STRING, b STRING, c STRING, PRIMARY KEY (a) NOT ENFORCED)"
+ + " WITH ('changelog-producer'='input', 'bucket' = '1')");
}
@Test
@@ -528,7 +529,7 @@ public class ContinuousFileStoreITCase extends CatalogITCaseBase {
public void testIgnoreDelete() {
sql(
"CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED, v STRING) "
- + "WITH ('deduplicate.ignore-delete' = 'true')");
+ + "WITH ('deduplicate.ignore-delete' = 'true', 'bucket' = '1')");
sql("INSERT INTO ignore_delete VALUES (1, 'A')");
assertThat(sql("SELECT * FROM ignore_delete")).containsExactly(Row.of(1, "A"));
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
index 00a47cd80..71f93553d 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
@@ -68,15 +68,15 @@ public class FileSystemCatalogITCase extends AbstractTestBase {
@Test
public void testWriteRead() throws Exception {
tEnv.useCatalog("fs");
- tEnv.executeSql("CREATE TABLE T (a STRING, b STRING, c STRING)");
+ tEnv.executeSql("CREATE TABLE T (a STRING, b STRING, c STRING) WITH ('bucket' = '1')");
innerTestWriteRead();
}
@Test
public void testRenameTable() throws Exception {
tEnv.useCatalog("fs");
- tEnv.executeSql("CREATE TABLE t1 (a INT)").await();
- tEnv.executeSql("CREATE TABLE t2 (a INT)").await();
+ tEnv.executeSql("CREATE TABLE t1 (a INT) WITH ('bucket' = '1')").await();
+ tEnv.executeSql("CREATE TABLE t2 (a INT) WITH ('bucket' = '1')").await();
tEnv.executeSql("INSERT INTO t1 VALUES(1),(2)").await();
// the source table do not exist.
assertThatThrownBy(() -> tEnv.executeSql("ALTER TABLE t3 RENAME TO t4"))
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
index aa9e41867..1ffa6fa7b 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
@@ -1838,6 +1838,7 @@ public class ReadWriteTableITCase extends AbstractTestBase {
options.put(SINK_PARALLELISM.key(), configParallelism.toString());
}
options.put("path", getTempFilePath(UUID.randomUUID().toString()));
+ options.put("bucket", "1");
DynamicTableFactory.Context context =
new FactoryUtil.DefaultDynamicTableContext(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
index a487ed4c1..143b4f396 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
@@ -96,7 +96,8 @@ public class RemoteLookupJoinITCase extends CatalogITCaseBase {
@Test
public void testLookupRemoteTable() throws Throwable {
- sql("CREATE TABLE DIM (i INT PRIMARY KEY NOT ENFORCED, j INT, k1 INT, k2 INT)");
+ sql(
+ "CREATE TABLE DIM (i INT PRIMARY KEY NOT ENFORCED, j INT, k1 INT, k2 INT) WITH ('bucket' = '1')");
ServiceProxy proxy = launchQueryServer("DIM");
proxy.write(GenericRow.of(1, 11, 111, 1111));
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
index 90d4235b2..af113e7c1 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
@@ -847,14 +847,15 @@ public class SchemaChangeITCase extends CatalogITCaseBase {
@Test
public void testSetAndResetImmutableOptions() throws Exception {
// bucket-key is immutable
- sql("CREATE TABLE T1 (a STRING, b STRING, c STRING)");
+ sql("CREATE TABLE T1 (a STRING, b STRING, c STRING) WITH ('bucket' = '1')");
assertThatThrownBy(() -> sql("ALTER TABLE T1 SET ('bucket-key' = 'c')"))
.getRootCause()
.isInstanceOf(UnsupportedOperationException.class)
.hasMessage("Change 'bucket-key' is not supported yet.");
- sql("CREATE TABLE T2 (a STRING, b STRING, c STRING) WITH ('bucket-key' = 'c')");
+ sql(
+ "CREATE TABLE T2 (a STRING, b STRING, c STRING) WITH ('bucket' = '1', 'bucket-key' = 'c')");
assertThatThrownBy(() -> sql("ALTER TABLE T2 RESET ('bucket-key')"))
.getRootCause()
.isInstanceOf(UnsupportedOperationException.class)
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java
index 476727699..80cf2cc76 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ActionITCaseBase.java
@@ -51,6 +51,7 @@ import org.junit.jupiter.api.BeforeEach;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -109,9 +110,13 @@ public abstract class ActionITCaseBase extends AbstractTestBase {
throws Exception {
Identifier identifier = Identifier.create(database, tableName);
catalog.createDatabase(database, true);
+ Map<String, String> newOptions = new HashMap<>(options);
+ if (!newOptions.containsKey("bucket")) {
+ newOptions.put("bucket", "1");
+ }
catalog.createTable(
identifier,
- new Schema(rowType.getFields(), partitionKeys, primaryKeys, options, ""),
+ new Schema(rowType.getFields(), partitionKeys, primaryKeys, newOptions, ""),
false);
return (FileStoreTable) catalog.getTable(identifier);
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
index 705826bfa..4697985d3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
@@ -93,6 +93,7 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
public void testBatchCompact(String mode) throws Exception {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ options.put("bucket", "1");
List<FileStoreTable> tables = new ArrayList<>();
@@ -174,6 +175,7 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
// test that dedicated compact job will expire snapshots
options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "3");
options.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "3");
+ options.put("bucket", "1");
List<FileStoreTable> tables = new ArrayList<>();
for (String dbName : DATABASE_NAMES) {
@@ -446,6 +448,7 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
throws Exception {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.WRITE_ONLY.key(), "true");
+ options.put("bucket", "1");
List<FileStoreTable> compactionTables = new ArrayList<>();
List<FileStoreTable> noCompactionTables = new ArrayList<>();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
index d9454f736..b01f5d082 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTestBase.java
@@ -94,6 +94,7 @@ public abstract class CommitterOperatorTestBase {
protected FileStoreTable createFileStoreTable(Consumer<Options> setOptions) throws Exception {
Options conf = new Options();
conf.set(CoreOptions.PATH, tablePath.toString());
+ conf.setString("bucket", "1");
setOptions.accept(conf);
SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(), tablePath);
schemaManager.createTable(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
index 8d5e6c781..9a71cc2ea 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CompactorSinkITCase.java
@@ -393,7 +393,7 @@ public class CompactorSinkITCase extends AbstractTestBase {
ROW_TYPE.getFields(),
Arrays.asList("dt", "hh"),
Arrays.asList("dt", "hh", "k"),
- Collections.emptyMap(),
+ Collections.singletonMap("bucket", "1"),
""));
return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
}
@@ -405,7 +405,7 @@ public class CompactorSinkITCase extends AbstractTestBase {
ROW_TYPE.getFields(),
Collections.emptyList(),
Collections.singletonList("k"),
- Collections.emptyMap(),
+ Collections.singletonMap("bucket", "1"),
"");
catalog.createTable(tableIdentifier, tableSchema, false);
return (FileStoreTable) catalog.getTable(tableIdentifier);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
index b214a4d32..608b659eb 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkSinkTest.java
@@ -62,7 +62,6 @@ import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -263,22 +262,23 @@ public class FlinkSinkTest {
private FileStoreTable createFileStoreTable() throws Exception {
org.apache.paimon.fs.Path tablePath = new org.apache.paimon.fs.Path(tempPath.toString());
- Options conf = new Options();
- conf.set(CoreOptions.PATH, tablePath.toString());
+ Options options = new Options();
+ options.set(CoreOptions.PATH, tablePath.toString());
+ options.set(CoreOptions.BUCKET, 1);
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(LocalFileIO.create(), tablePath),
new Schema(
ROW_TYPE.getFields(),
Collections.emptyList(),
- Arrays.asList("pk"),
- conf.toMap(),
+ Collections.singletonList("pk"),
+ options.toMap(),
""));
return FileStoreTableFactory.create(
FileIOFinder.find(tablePath),
tablePath,
tableSchema,
- conf,
+ options,
new CatalogEnvironment(Lock.emptyFactory(), null, null));
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PrimaryKeyWriterOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PrimaryKeyWriterOperatorTest.java
index 907c2d4cb..c1e623fe0 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PrimaryKeyWriterOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/PrimaryKeyWriterOperatorTest.java
@@ -25,6 +25,7 @@ public class PrimaryKeyWriterOperatorTest extends WriterOperatorTestBase {
@Override
protected void setTableConfig(Options options) {
options.set("primary-key", "a");
+ options.set("bucket", "1");
options.set("bucket-key", "a");
options.set("write-buffer-size", "256 b");
options.set("page-size", "32 b");
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/RowDataChannelComputerTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/RowDataChannelComputerTest.java
index f727cf981..f90a3cabb 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/RowDataChannelComputerTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/RowDataChannelComputerTest.java
@@ -66,7 +66,7 @@ public class RowDataChannelComputerTest {
rowType.getFields(),
Collections.singletonList("pt"),
Arrays.asList("pt", "k"),
- new HashMap<>(),
+ Collections.singletonMap("bucket", "1"),
""));
ThreadLocalRandom random = ThreadLocalRandom.current();
@@ -95,7 +95,7 @@ public class RowDataChannelComputerTest {
rowType.getFields(),
Collections.emptyList(),
Collections.singletonList("k"),
- new HashMap<>(),
+ Collections.singletonMap("bucket", "1"),
""));
ThreadLocalRandom random = ThreadLocalRandom.current();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
index 9d0a15fe2..3730a1930 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java
@@ -119,6 +119,7 @@ class StoreMultiCommitterTest {
Options firstOptions = new Options();
firstOptions.set(
CoreOptions.TAG_AUTOMATIC_CREATION, CoreOptions.TagCreationMode.PROCESS_TIME);
+ firstOptions.setString("bucket", "-1");
Schema firstTableSchema =
new Schema(
rowType1.getFields(),
@@ -132,7 +133,7 @@ class StoreMultiCommitterTest {
rowType2.getFields(),
Collections.emptyList(),
Collections.emptyList(),
- Collections.emptyMap(),
+ Collections.singletonMap("bucket", "1"),
"");
createTestTables(
catalog,
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java
index b72729594..2a95b5d58 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java
@@ -318,7 +318,7 @@ public class CompactorSourceITCase extends AbstractTestBase {
ROW_TYPE.getFields(),
Arrays.asList("dt", "hh"),
Arrays.asList("dt", "hh", "k"),
- Collections.emptyMap(),
+ Collections.singletonMap("bucket", "1"),
""));
return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java
index e83117735..91af8071e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java
@@ -54,6 +54,7 @@ import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
/** Tests for {@link DataTableSource}. */
class DataTableSourceTest {
+
@Test
void testInferScanParallelism(@TempDir java.nio.file.Path path) throws Exception {
FileIO fileIO = LocalFileIO.create();
@@ -64,6 +65,7 @@ class DataTableSourceTest {
Schema.newBuilder()
.column("a", DataTypes.INT())
.column("b", DataTypes.BIGINT())
+ .option("bucket", "1")
.build());
FileStoreTable fileStoreTable =
FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
index 6b998c43e..56e9374d5 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MultiTablesCompactorSourceBuilderITCase.java
@@ -112,6 +112,7 @@ public class MultiTablesCompactorSourceBuilderITCase extends AbstractTestBase
// change options to test whether CompactorSourceBuilder work normally
options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), "2");
}
+ options.put("bucket", "1");
long monitorInterval = 1000;
for (String dbName : DATABASE_NAMES) {
@@ -203,6 +204,7 @@ public class MultiTablesCompactorSourceBuilderITCase extends AbstractTestBase
CoreOptions.ChangelogProducer.NONE.toString());
options.put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), "0");
}
+ options.put("bucket", "1");
long monitorInterval = 1000;
List<FileStoreTable> tables = new ArrayList<>();
for (String dbName : DATABASE_NAMES) {
@@ -372,6 +374,7 @@ public class MultiTablesCompactorSourceBuilderITCase extends AbstractTestBase
CoreOptions.ChangelogProducer.NONE.toString());
options.put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), "0");
}
+ options.put("bucket", "1");
long monitorInterval = 1000;
for (String dbName : DATABASE_NAMES) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
index f6e6d778d..3c046457e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
@@ -55,7 +55,6 @@ import java.io.IOException;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -83,7 +82,8 @@ public class OperatorSourceTest {
.column("b", DataTypes.INT())
.column("c", DataTypes.INT())
.primaryKey("a")
- .options(Collections.singletonMap(CONSUMER_ID.key(), "my_consumer"))
+ .option(CONSUMER_ID.key(), "my_consumer")
+ .option("bucket", "1")
.build();
Identifier identifier = Identifier.create("default", "t");
catalog.createDatabase("default", false);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/PrimaryKeyTableStatisticsTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/PrimaryKeyTableStatisticsTest.java
index 91a607aaf..8e4643ef5 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/PrimaryKeyTableStatisticsTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/PrimaryKeyTableStatisticsTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.flink.source.DataTableSource;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
@@ -57,13 +58,14 @@ public class PrimaryKeyTableStatisticsTest extends FileStoreTableStatisticsTestB
@Override
FileStoreTable createStoreTable() throws Exception {
- Options conf = new Options();
- conf.set(CoreOptions.PATH, tablePath.toString());
- conf.set(CoreOptions.BUCKET, 1);
+ Options options = new Options();
+ options.set(CoreOptions.PATH, tablePath.toString());
+ options.set(CoreOptions.BUCKET, 1);
+ Schema.Builder builder = schemaBuilder();
+ builder.options(options.toMap());
TableSchema tableSchema =
new SchemaManager(LocalFileIO.create(), tablePath)
- .createTable(
- schemaBuilder().partitionKeys("pt").primaryKey("pt", "a").build());
+ .createTable(builder.partitionKeys("pt").primaryKey("pt", "a").build());
return FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java
index 44783009d..2303ba1bc 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/util/ReadWriteTableTestUtil.java
@@ -121,7 +121,11 @@ public class ReadWriteTableTestUtil {
Map<String, String> options) {
// "-" is not allowed in the table name.
String table = ("MyTable_" + UUID.randomUUID()).replace("-", "_");
- sEnv.executeSql(buildDdl(table, fieldsSpec, primaryKeys, partitionKeys, options));
+ Map<String, String> newOptions = new HashMap<>(options);
+ if (!newOptions.containsKey("bucket")) {
+ newOptions.put("bucket", "1");
+ }
+ sEnv.executeSql(buildDdl(table, fieldsSpec, primaryKeys, partitionKeys, newOptions));
return table;
}
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java
index d057cb52d..2830198e9 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java
@@ -591,6 +591,7 @@ public class HiveWriteITCase extends HiveTestBase {
Options conf = new Options();
conf.set(CatalogOptions.WAREHOUSE, root);
conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set("bucket", "1");
Table table =
FileStoreTestUtils.createFileStoreTable(
conf,
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/mapred/PaimonRecordReaderTest.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/mapred/PaimonRecordReaderTest.java
index 738958869..8597a84dc 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/mapred/PaimonRecordReaderTest.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/mapred/PaimonRecordReaderTest.java
@@ -62,6 +62,7 @@ public class PaimonRecordReaderTest {
Options conf = new Options();
conf.set(CatalogOptions.WAREHOUSE, tempDir.toString());
conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set("bucket", "1");
Table table =
FileStoreTestUtils.createFileStoreTable(
conf,
diff --git a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java
index 6e7c5effa..72691a41d 100644
--- a/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java
+++ b/paimon-service/paimon-service-runtime/src/test/java/org/apache/paimon/service/KvQueryTableTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.catalog.PrimaryKeyTableTestBase;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.options.Options;
import org.apache.paimon.query.QueryLocationImpl;
import org.apache.paimon.service.client.KvQueryClient;
import org.apache.paimon.service.network.stats.DisabledServiceRequestStats;
@@ -57,6 +58,13 @@ public class KvQueryTableTest extends PrimaryKeyTableTestBase {
private KvQueryClient client;
+ @Override
+ protected Options tableOptions() {
+ Options options = new Options();
+ options.set("bucket", "1");
+ return options;
+ }
+
@BeforeEach
public void beforeEach() {
IOManager ioManager = IOManager.create(tempPath.toString());
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java
index b04a3c912..99066dcfd 100644
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java
+++ b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java
@@ -48,6 +48,7 @@ public class SimpleTableTestHelper {
Map<String, String> options = new HashMap<>();
// orc is shaded, can not find shaded classes in ide
options.put(CoreOptions.FILE_FORMAT.key(), CoreOptions.FileFormatType.AVRO.toString());
+ options.put("bucket", "1");
new SchemaManager(LocalFileIO.create(), path)
.createTable(
new Schema(
diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkReadTestBase.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkReadTestBase.java
index ba421d09a..29ab2ff65 100644
--- a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkReadTestBase.java
+++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkReadTestBase.java
@@ -164,7 +164,7 @@ public abstract class SparkReadTestBase {
protected static void createTable(String tableName) {
spark.sql(
String.format(
- "CREATE TABLE paimon.default.%s (a INT NOT NULL, b BIGINT, c STRING) TBLPROPERTIES ('primary-key'='a', 'file.format'='avro')",
+ "CREATE TABLE paimon.default.%s (a INT NOT NULL, b BIGINT, c STRING) TBLPROPERTIES ('bucket' = '1', 'primary-key'='a', 'file.format'='avro')",
tableName));
}
diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelITCase.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelITCase.java
index 2c31664c2..bc77da589 100644
--- a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelITCase.java
+++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkTimeTravelITCase.java
@@ -168,7 +168,7 @@ public class SparkTimeTravelITCase extends SparkReadTestBase {
@Test
public void testSystemTableTimeTravel() throws Exception {
- spark.sql("CREATE TABLE t (k INT, v STRING)");
+ spark.sql("CREATE TABLE t (k INT, v STRING) TBLPROPERTIES ('bucket' = '1')");
// snapshot 1
writeTable(
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala
index 3a76503f1..54653e9f9 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DeleteFromTableTest.scala
@@ -62,10 +62,11 @@ class DeleteFromTableTest extends PaimonSparkTestBase {
}
test(s"test delete with primary key") {
- spark.sql(s"""
- |CREATE TABLE T (id INT, name STRING, dt STRING)
- |TBLPROPERTIES ('primary-key' = 'id', 'merge-engine' = 'deduplicate')
- |""".stripMargin)
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, name STRING, dt STRING)
+ |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1', 'merge-engine' = 'deduplicate')
+ |""".stripMargin)
spark.sql("INSERT INTO T VALUES (1, 'a', '11'), (2, 'b', '22'), (3, 'c', '33')")
@@ -81,10 +82,11 @@ class DeleteFromTableTest extends PaimonSparkTestBase {
}
test(s"test delete with non-primary key") {
- spark.sql(s"""
- |CREATE TABLE T (id INT, name STRING, dt STRING)
- |TBLPROPERTIES ('primary-key' = 'id', 'merge-engine' = 'deduplicate')
- |""".stripMargin)
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, name STRING, dt STRING)
+ |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1', 'merge-engine' = 'deduplicate')
+ |""".stripMargin)
spark.sql("INSERT INTO T VALUES (1, 'a', '11'), (2, 'b', '22'), (3, 'c', '33'), (4, 'a', '44')")
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
index 012e69cf1..4727df2cc 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
@@ -39,7 +39,7 @@ class LookupCompactionTest extends PaimonSparkTestBase {
spark.sql(
s"""
|CREATE TABLE T (id INT, name STRING, count INT)
- |TBLPROPERTIES ('primary-key' = 'id', 'merge-engine' = '$mergeEngine', 'changelog-producer' = 'lookup' $extraOptions)
+ |TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1', 'merge-engine' = '$mergeEngine', 'changelog-producer' = 'lookup' $extraOptions)
|""".stripMargin)
val table = loadTable("T")