You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2023/06/09 03:31:22 UTC
[incubator-paimon] branch master updated: [core] Add unaware-bucket process for 'AppendOnlyFileStoreTable', add test for unaware-bucket table check and compaction (#1337)
This is an automated email from the ASF dual-hosted git repository.
yuzelin 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 4e25074eb [core] Add unaware-bucket process for 'AppendOnlyFileStoreTable', add test for unaware-bucket table check and compaction (#1337)
4e25074eb is described below
commit 4e25074eb5a726ebdbf64e7b5cc50a3bf8b38b09
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Fri Jun 9 11:31:18 2023 +0800
[core] Add unaware-bucket process for 'AppendOnlyFileStoreTable', add test for unaware-bucket table check and compaction (#1337)
---
.../main/java/org/apache/paimon/CoreOptions.java | 6 +-
.../AppendOnlyTableCompactionCoordinator.java | 15 +-
.../org/apache/paimon/schema/SchemaValidation.java | 6 +
.../paimon/table/AbstractFileStoreTable.java | 5 +
.../paimon/table/AppendOnlyFileStoreTable.java | 9 +-
.../table/sink/UnawareBucketRowKeyExtractor.java | 42 +++++
.../table/source/AbstractInnerTableScan.java | 2 +-
.../table/source/InnerStreamTableScanImpl.java | 2 +-
.../append/AppendOnlyTableCompactionITTest.java | 189 +++++++++++++++++++++
.../sink/UnawareBucketRowKeyExtractorTest.java | 76 +++++++++
.../flink/source/CompactorSourceBuilder.java | 4 +-
.../apache/paimon/flink/AppendOnlyTableITCase.java | 12 ++
12 files changed, 360 insertions(+), 8 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
index bb3b2eac5..0d66e95e0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1203,9 +1203,9 @@ public class CoreOptions implements Serializable {
/** Compaction type when trigger a compaction action. */
public enum StreamingCompactionType implements DescribedEnum {
- NONE("none", "not a streaming compaction."),
- NORMAL("normal", "Traditional bucket table compaction."),
- NON_BUCKET("non-bucket", "Compaction for non-bucket table.");
+ NONE("none", "Not a streaming compaction."),
+ NORMAL("normal", "Compaction for traditional bucket table."),
+ BUCKET_UNAWARE("unaware", "Compaction for unaware bucket table.");
private final String value;
private final String description;
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java
index e1b0c1f32..4f125ddd9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinator.java
@@ -126,10 +126,23 @@ public class AppendOnlyTableCompactionCoordinator {
return tasks;
}
+ @VisibleForTesting
+ HashSet<DataFileMeta> listRestoredFiles() {
+ HashSet<DataFileMeta> sets = new HashSet<>();
+ partitionCompactCoordinators
+ .values()
+ .forEach(
+ partitionCompactCoordinator ->
+ sets.addAll(partitionCompactCoordinator.toCompact));
+ return sets;
+ }
+
private Map<String, String> compactScanType() {
return new HashMap<String, String>() {
{
- put(CoreOptions.STREAMING_COMPACT.key(), "non-bucket");
+ put(
+ CoreOptions.STREAMING_COMPACT.key(),
+ CoreOptions.StreamingCompactionType.BUCKET_UNAWARE.getValue());
}
};
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index b2370e5c2..ee7b1a08f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -37,6 +37,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import static org.apache.paimon.CoreOptions.BUCKET_KEY;
import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
import static org.apache.paimon.CoreOptions.SCAN_MODE;
import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
@@ -144,6 +145,11 @@ public class SchemaValidation {
+ "still want to keep the primary key definition.");
}
+ if (options.bucket() == -1 && options.toMap().get(BUCKET_KEY.key()) != null) {
+ throw new RuntimeException(
+ "Cannot define 'bucket-key' in unaware or dynamic bucket mode.");
+ }
+
if (schema.primaryKeys().isEmpty() && options.streamingReadOverwrite()) {
throw new RuntimeException(
"Doesn't support streaming read the changes from overwrite when the primary keys are not defined.");
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 cbb81815b..7f2c9d602 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
@@ -32,6 +32,7 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
import org.apache.paimon.table.sink.RowKeyExtractor;
import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.UnawareBucketRowKeyExtractor;
import org.apache.paimon.table.source.InnerStreamTableScan;
import org.apache.paimon.table.source.InnerStreamTableScanImpl;
import org.apache.paimon.table.source.InnerTableScan;
@@ -87,7 +88,11 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
case FIXED:
return new FixedBucketRowKeyExtractor(schema());
case DYNAMIC:
+ // todo: dynamic bucket table need to check current bucket, when full, return the
+ // next one.
+ throw new UnsupportedOperationException("Not supported yet.");
case UNAWARE:
+ return new UnawareBucketRowKeyExtractor(schema());
default:
throw new UnsupportedOperationException("Unsupported mode: " + bucketMode());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
index 6c1110cae..cd342db5f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
@@ -27,6 +27,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ManifestCacheFilter;
import org.apache.paimon.operation.AppendOnlyFileStoreRead;
import org.apache.paimon.operation.AppendOnlyFileStoreScan;
+import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
import org.apache.paimon.operation.FileStoreScan;
import org.apache.paimon.operation.ReverseReader;
import org.apache.paimon.predicate.Predicate;
@@ -129,8 +130,14 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWriteImpl<InternalRow> newWrite(
String commitUser, ManifestCacheFilter manifestFilter) {
+ AppendOnlyFileStoreWrite writer = store().newWrite(commitUser, manifestFilter);
+ // if this table is non-bucket table, we skip compaction and restored files searching
+ if (bucketMode() == BucketMode.UNAWARE) {
+ writer.skipCompaction();
+ writer.fromEmptyWriter(true);
+ }
return new TableWriteImpl<>(
- store().newWrite(commitUser, manifestFilter),
+ writer,
createRowKeyExtractor(),
record -> {
Preconditions.checkState(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java
new file mode 100644
index 000000000..e99133899
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractor.java
@@ -0,0 +1,42 @@
+/*
+ * 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.sink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.schema.TableSchema;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Unaware bucket table put all records into bucket 0 (for comparability with old design). */
+public class UnawareBucketRowKeyExtractor extends RowKeyExtractor {
+
+ public UnawareBucketRowKeyExtractor(TableSchema schema) {
+ super(schema);
+ int numBuckets = new CoreOptions(schema.options()).bucket();
+ checkArgument(
+ numBuckets == -1,
+ "Only 'bucket' = '-1' is allowed for 'UnawareBucketRowKeyExtractor', but found: "
+ + numBuckets);
+ }
+
+ @Override
+ public int bucket() {
+ return 0;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
index f8e03df89..35f0f5c1c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractInnerTableScan.java
@@ -75,7 +75,7 @@ public abstract class AbstractInnerTableScan implements InnerTableScan {
"Set 'streaming-compact' in batch mode. This is unexpected.");
return new ContinuousCompactorStartingScanner();
}
- case NON_BUCKET:
+ case BUCKET_UNAWARE:
{
return new FullStartingScanner();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
index 8bc051cf6..42f8714f7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerStreamTableScanImpl.java
@@ -165,7 +165,7 @@ public class InnerStreamTableScanImpl extends AbstractInnerTableScan
{
return new ContinuousCompactorFollowUpScanner();
}
- case NON_BUCKET:
+ case BUCKET_UNAWARE:
{
return new ContinuousAppendAndCompactFollowUpScanner();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionITTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionITTest.java
new file mode 100644
index 000000000..6e2df6572
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionITTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.append;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+
+/** Test for {@link AppendOnlyTableCompactionCoordinator}. */
+public class AppendOnlyTableCompactionITTest {
+
+ @TempDir private Path tempDir;
+ private AppendOnlyFileStoreTable appendOnlyFileStoreTable;
+ private SnapshotManager snapshotManager;
+ private AppendOnlyTableCompactionCoordinator compactionCoordinator;
+ private AppendOnlyTableCompactionWorker compactionWorker;
+ private final String commitUser = UUID.randomUUID().toString();
+
+ @Test
+ public void noCompaction() throws Exception {
+ List<CommitMessage> messages = writeCommit(10);
+
+ messages.forEach(
+ message ->
+ Assertions.assertTrue(
+ ((CommitMessageImpl) message).compactIncrement().isEmpty()));
+ }
+
+ @Test
+ public void compactionTaskTest() throws Exception {
+ // commit 11 files
+ List<CommitMessage> messages = writeCommit(11);
+ commit(messages);
+
+ // first compact, six files left after commit compact and update restored files
+ // test run method invoke scan and compactPlan
+ List<AppendOnlyCompactionTask> tasks = compactionCoordinator.run();
+ Assertions.assertEquals(tasks.size(), 1);
+ AppendOnlyCompactionTask task = tasks.get(0);
+ Assertions.assertEquals(task.compactBefore().size(), 6);
+ compactionWorker.accept(tasks);
+ List<CommitMessage> result = compactionWorker.doCompact();
+ Assertions.assertEquals(1, result.size());
+ commit(result);
+ compactionCoordinator.scan();
+ Assertions.assertEquals(compactionCoordinator.listRestoredFiles().size(), 6);
+
+ // second compact, only one file left after updateRestored
+ tasks = compactionCoordinator.compactPlan();
+ Assertions.assertEquals(tasks.size(), 1);
+ // before update, zero file left
+ Assertions.assertEquals(compactionCoordinator.listRestoredFiles().size(), 0);
+ commit(compactionWorker.accept(tasks).doCompact());
+ compactionCoordinator.scan();
+ // one file is loaded from delta
+ List<DataFileMeta> last = new ArrayList<>(compactionCoordinator.listRestoredFiles());
+ Assertions.assertEquals(last.size(), 1);
+ Assertions.assertEquals(last.get(0).rowCount(), 11);
+ }
+
+ @Test
+ public void testCompactionLot() throws Exception {
+ // test continuous compaction
+ Assertions.assertNull(snapshotManager.latestSnapshotId());
+
+ long count = 0;
+ for (int i = 90; i < 100; i++) {
+ count += i;
+ commit(writeCommit(i));
+ commit(compactionWorker.accept(compactionCoordinator.run()).doCompact());
+ // scan the file generated itself
+ Assertions.assertTrue(compactionCoordinator.scan());
+ Assertions.assertEquals(
+ compactionCoordinator.listRestoredFiles().stream()
+ .map(DataFileMeta::rowCount)
+ .reduce(Long::sum)
+ .get(),
+ count);
+ }
+
+ Assertions.assertEquals(
+ appendOnlyFileStoreTable.store().newScan().plan().files().size(),
+ compactionCoordinator.listRestoredFiles().size());
+
+ List<AppendOnlyCompactionTask> tasks = compactionCoordinator.compactPlan();
+ while (tasks.size() != 0) {
+ commit(compactionWorker.accept(tasks).doCompact());
+ tasks = compactionCoordinator.run();
+ }
+
+ int remainedSize = appendOnlyFileStoreTable.store().newScan().plan().files().size();
+ Assertions.assertEquals(remainedSize, compactionCoordinator.listRestoredFiles().size());
+ Assertions.assertEquals(remainedSize, 5);
+ }
+
+ private static Schema schema() {
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ schemaBuilder.column("f0", DataTypes.INT());
+ schemaBuilder.column("f1", DataTypes.STRING());
+ schemaBuilder.column("f2", DataTypes.STRING());
+ schemaBuilder.column("f3", DataTypes.STRING());
+ schemaBuilder.option("compaction.min.file-num", "3");
+ schemaBuilder.option("compaction.max.file-num", "6");
+ schemaBuilder.option("bucket", "-1");
+ return schemaBuilder.build();
+ }
+
+ private void commit(List<CommitMessage> messages) {
+ appendOnlyFileStoreTable.newCommit(commitUser).commit(messages);
+ }
+
+ private List<CommitMessage> writeCommit(int number) throws Exception {
+ List<CommitMessage> messages = new ArrayList<>();
+ StreamTableWrite writer = appendOnlyFileStoreTable.newStreamWriteBuilder().newWrite();
+ for (int i = 0; i < number; i++) {
+ writer.write(randomRow());
+ messages.addAll(writer.prepareCommit(true, i));
+ }
+ return messages;
+ }
+
+ private InternalRow randomRow() {
+ return GenericRow.of(
+ random.nextInt(100),
+ BinaryString.fromString("A" + random.nextInt(100)),
+ BinaryString.fromString("B" + random.nextInt(100)),
+ BinaryString.fromString("C" + random.nextInt(100)));
+ }
+
+ private static final Random random = new Random();
+
+ @BeforeEach
+ public void createNegativeAppendOnlyTable() throws Exception {
+ FileIO fileIO = new LocalFileIO();
+ org.apache.paimon.fs.Path path = new org.apache.paimon.fs.Path(tempDir.toString());
+ SchemaManager schemaManager = new SchemaManager(fileIO, path);
+ TableSchema tableSchema = schemaManager.createTable(schema());
+ snapshotManager = new SnapshotManager(fileIO, path);
+ appendOnlyFileStoreTable =
+ (AppendOnlyFileStoreTable)
+ FileStoreTableFactory.create(
+ fileIO,
+ new org.apache.paimon.fs.Path(tempDir.toString()),
+ tableSchema);
+ compactionCoordinator = new AppendOnlyTableCompactionCoordinator(appendOnlyFileStoreTable);
+ compactionWorker =
+ new AppendOnlyTableCompactionWorker(appendOnlyFileStoreTable, commitUser);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java
new file mode 100644
index 000000000..0ac286148
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/UnawareBucketRowKeyExtractorTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.sink;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.BUCKET;
+
+/** Test for {@link UnawareBucketRowKeyExtractor}. */
+public class UnawareBucketRowKeyExtractorTest {
+
+ @Test
+ public void testBucket() {
+ GenericRow row = GenericRow.of(5, 6, 7);
+ Assertions.assertEquals(bucket(extractor("a"), row), 0);
+ }
+
+ private int bucket(UnawareBucketRowKeyExtractor extractor, InternalRow row) {
+ extractor.setRecord(row);
+ return extractor.bucket();
+ }
+
+ private UnawareBucketRowKeyExtractor extractor(String partK) {
+ RowType rowType =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "a", new IntType()),
+ new DataField(1, "b", new IntType()),
+ new DataField(2, "c", new IntType())));
+ List<DataField> fields = TableSchema.newFields(rowType);
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "-1");
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ fields,
+ RowType.currentHighestFieldId(fields),
+ "".equals(partK)
+ ? Collections.emptyList()
+ : Arrays.asList(partK.split(",")),
+ Collections.emptyList(),
+ options,
+ "");
+ return new UnawareBucketRowKeyExtractor(schema);
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
index 755e878bc..3a29d3521 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -129,7 +129,9 @@ public class CompactorSourceBuilder {
// set 'streaming-compact' and remove 'scan.bounded.watermark'
return new HashMap<String, String>() {
{
- put(CoreOptions.STREAMING_COMPACT.key(), "normal");
+ put(
+ CoreOptions.STREAMING_COMPACT.key(),
+ CoreOptions.StreamingCompactionType.NORMAL.getValue());
put(CoreOptions.SCAN_BOUNDED_WATERMARK.key(), null);
}
};
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 d295b1013..f67f64edf 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
@@ -50,6 +50,18 @@ public class AppendOnlyTableITCase extends CatalogITCaseBase {
+ "want to keep the primary key definition.");
}
+ @Test
+ public void testCreateNonBucketTableWithBucketKey() {
+ assertThatThrownBy(
+ () ->
+ batchSql(
+ "CREATE TABLE pk_table (id INT, data STRING) "
+ + "WITH ('bucket' = '-1', 'bucket-key' = 'id')"))
+ .hasRootCauseInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage(
+ "Cannot define 'bucket-key' in unaware or dynamic bucket mode.");
+ }
+
@Test
public void testReadEmpty() {
assertThat(batchSql("SELECT * FROM append_table")).isEmpty();