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 437d421d0 [core] Make PrimaryKeyFileStoreTable and
AppendOnlyFileStoreTable package private (#2791)
437d421d0 is described below
commit 437d421d0290f07fbbd01adf118a0e660b5bb66f
Author: tsreaper <[email protected]>
AuthorDate: Thu Jan 25 16:32:11 2024 +0800
[core] Make PrimaryKeyFileStoreTable and AppendOnlyFileStoreTable package
private (#2791)
---
.../AppendOnlyTableCompactionCoordinator.java | 25 ++++++-----
.../paimon/table/AppendOnlyFileStoreTable.java | 2 +-
.../paimon/table/PrimaryKeyFileStoreTable.java | 2 +-
.../AppendOnlyTableCompactionCoordinatorTest.java | 11 ++---
.../append/AppendOnlyTableCompactionITTest.java | 13 +++---
.../operation/AppendOnlyFileStoreWriteTest.java | 10 ++---
.../org/apache/paimon/table/TableTestBase.java | 4 +-
.../paimon/flink/sink/cdc/CdcSinkBuilder.java | 4 +-
.../flink/sink/cdc/CdcUnawareBucketSink.java | 4 +-
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 3 +-
.../apache/paimon/flink/action/CompactAction.java | 6 +--
.../paimon/flink/action/CompactDatabaseAction.java | 25 +++++------
.../UnawareBucketCompactionTopoBuilder.java | 8 ++--
.../AppendOnlyTableCompactionWorkerOperator.java | 9 ++--
.../apache/paimon/flink/sink/FlinkSinkBuilder.java | 9 +---
.../paimon/flink/sink/FlinkTableSinkBase.java | 9 +---
.../paimon/flink/sink/RowUnawareBucketSink.java | 4 +-
.../SupportsRowLevelOperationFlinkTableSink.java | 20 ++-------
.../flink/sink/UnawareBucketCompactionSink.java | 8 ++--
.../paimon/flink/sink/UnawareBucketSink.java | 6 +--
.../flink/source/BucketUnawareCompactSource.java | 6 +--
.../paimon/flink/source/DataTableSource.java | 11 +----
.../SortCompactActionForDynamicBucketITCase.java | 48 ++++++++--------------
.../SortCompactActionForUnawareBucketITCase.java | 44 ++++++++------------
...ppendOnlyTableCompactionWorkerOperatorTest.java | 9 ++--
.../statistics/AppendOnlyTableStatisticsTest.java | 3 +-
.../statistics/PrimaryKeyTableStatisticsTest.java | 5 +--
.../apache/paimon/hive/migrate/HiveMigrator.java | 3 +-
.../paimon/spark/procedure/CompactProcedure.java | 7 ++--
.../apache/paimon/spark/PaimonScanBuilder.scala | 5 +--
.../spark/catalyst/analysis/RowLevelHelper.scala | 15 +++----
31 files changed, 126 insertions(+), 212 deletions(-)
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 e37d57336..614544ff1 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
@@ -23,11 +23,11 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.Preconditions;
import javax.annotation.Nullable;
@@ -41,14 +41,14 @@ import java.util.Map;
import java.util.stream.Collectors;
/**
- * {@link AppendOnlyFileStoreTable} compact coordinator.
+ * Compact coordinator for append only tables.
*
- * <p>Note: AppendOnlyTableCompactionCoordinator scan files in snapshot, read
APPEND and COMPACT
- * snapshot then load those new files. It will try it best to generate
compaction task for the
- * restored files scanned in snapshot, but to reduce memory usage, it won't
remain single file for a
- * long time. After ten times scan, single file with one partition will be
ignored and removed from
- * memory, which means, it will not participate in compaction again until
restart the compaction
- * job.
+ * <p>Note: {@link AppendOnlyTableCompactionCoordinator} scan files in
snapshot, read APPEND and
+ * COMPACT snapshot then load those new files. It will try it best to generate
compaction task for
+ * the restored files scanned in snapshot, but to reduce memory usage, it
won't remain single file
+ * for a long time. After ten times scan, single file with one partition will
be ignored and removed
+ * from memory, which means, it will not participate in compaction again until
restart the
+ * compaction job.
*
* <p>When a third task delete file in latest snapshot(including batch
delete/update and overwrite),
* the file in coordinator will still remain and participate in compaction
task. When this happens,
@@ -70,18 +70,17 @@ public class AppendOnlyTableCompactionCoordinator {
final Map<BinaryRow, PartitionCompactCoordinator>
partitionCompactCoordinators =
new HashMap<>();
- public AppendOnlyTableCompactionCoordinator(AppendOnlyFileStoreTable
table) {
+ public AppendOnlyTableCompactionCoordinator(FileStoreTable table) {
this(table, true);
}
- public AppendOnlyTableCompactionCoordinator(
- AppendOnlyFileStoreTable table, boolean isStreaming) {
+ public AppendOnlyTableCompactionCoordinator(FileStoreTable table, boolean
isStreaming) {
this(table, isStreaming, null);
}
public AppendOnlyTableCompactionCoordinator(
- AppendOnlyFileStoreTable table, boolean isStreaming, @Nullable
Predicate filter) {
-
+ FileStoreTable table, boolean isStreaming, @Nullable Predicate
filter) {
+ Preconditions.checkArgument(table.primaryKeys().isEmpty());
FileStoreTable tableCopy = table.copy(compactScanType());
if (isStreaming) {
scan = tableCopy.newStreamScan();
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 678453729..915ab1544 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
@@ -47,7 +47,7 @@ import java.io.IOException;
import java.util.function.BiConsumer;
/** {@link FileStoreTable} for append table. */
-public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
+class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
private static final long serialVersionUID = 1L;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
index d815aae9f..a1614765e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
@@ -56,7 +56,7 @@ import static
org.apache.paimon.predicate.PredicateBuilder.pickTransformFieldMap
import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
/** {@link FileStoreTable} for primary key table. */
-public class PrimaryKeyFileStoreTable extends AbstractFileStoreTable {
+class PrimaryKeyFileStoreTable extends AbstractFileStoreTable {
private static final long serialVersionUID = 1L;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinatorTest.java
index 83e5be14c..0b00cfefa 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinatorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionCoordinatorTest.java
@@ -26,7 +26,7 @@ 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.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.types.DataTypes;
@@ -48,7 +48,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class AppendOnlyTableCompactionCoordinatorTest {
@TempDir Path tempDir;
- private AppendOnlyFileStoreTable appendOnlyFileStoreTable;
+ private FileStoreTable appendOnlyFileStoreTable;
private AppendOnlyTableCompactionCoordinator compactionCoordinator;
private BinaryRow partition;
@@ -159,11 +159,8 @@ public class AppendOnlyTableCompactionCoordinatorTest {
TableSchema tableSchema = schemaManager.createTable(schema());
appendOnlyFileStoreTable =
- (AppendOnlyFileStoreTable)
- FileStoreTableFactory.create(
- fileIO,
- new
org.apache.paimon.fs.Path(tempDir.toString()),
- tableSchema);
+ FileStoreTableFactory.create(
+ fileIO, new
org.apache.paimon.fs.Path(tempDir.toString()), tableSchema);
compactionCoordinator = new
AppendOnlyTableCompactionCoordinator(appendOnlyFileStoreTable);
partition = BinaryRow.EMPTY_ROW;
}
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
index f0a13d9fa..a67a081a0 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionITTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyTableCompactionITTest.java
@@ -28,7 +28,7 @@ import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
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.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -53,7 +53,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class AppendOnlyTableCompactionITTest {
@TempDir private Path tempDir;
- private AppendOnlyFileStoreTable appendOnlyFileStoreTable;
+ private FileStoreTable appendOnlyFileStoreTable;
private SnapshotManager snapshotManager;
private AppendOnlyTableCompactionCoordinator compactionCoordinator;
private AppendOnlyFileStoreWrite write;
@@ -188,12 +188,9 @@ public class AppendOnlyTableCompactionITTest {
TableSchema tableSchema = schemaManager.createTable(schema());
snapshotManager = new SnapshotManager(fileIO, path);
appendOnlyFileStoreTable =
- (AppendOnlyFileStoreTable)
- FileStoreTableFactory.create(
- fileIO,
- new
org.apache.paimon.fs.Path(tempDir.toString()),
- tableSchema);
+ FileStoreTableFactory.create(
+ fileIO, new
org.apache.paimon.fs.Path(tempDir.toString()), tableSchema);
compactionCoordinator = new
AppendOnlyTableCompactionCoordinator(appendOnlyFileStoreTable);
- write = appendOnlyFileStoreTable.store().newWrite(commitUser);
+ write = (AppendOnlyFileStoreWrite)
appendOnlyFileStoreTable.store().newWrite(commitUser);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
index 661a2a76c..9281a7b35 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/AppendOnlyFileStoreWriteTest.java
@@ -31,7 +31,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.types.DataTypes;
@@ -50,9 +50,9 @@ public class AppendOnlyFileStoreWriteTest {
@Test
public void testWritesInBatch() throws Exception {
- AppendOnlyFileStoreTable table = createFileStoreTable();
+ FileStoreTable table = createFileStoreTable();
- AppendOnlyFileStoreWrite write = table.store().newWrite("ss");
+ AppendOnlyFileStoreWrite write = (AppendOnlyFileStoreWrite)
table.store().newWrite("ss");
write.withExecutionMode(false);
write.write(partition(0), 0, GenericRow.of(0, 0, 0));
@@ -101,7 +101,7 @@ public class AppendOnlyFileStoreWriteTest {
Assertions.assertThat(records).isEqualTo(11);
}
- protected AppendOnlyFileStoreTable createFileStoreTable() throws Exception
{
+ protected FileStoreTable createFileStoreTable() throws Exception {
Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new
Path(tempDir.toString()));
Schema schema =
Schema.newBuilder()
@@ -114,7 +114,7 @@ public class AppendOnlyFileStoreWriteTest {
Identifier identifier = Identifier.create("default", "test");
catalog.createDatabase("default", false);
catalog.createTable(identifier, schema, false);
- return (AppendOnlyFileStoreTable) catalog.getTable(identifier);
+ return (FileStoreTable) catalog.getTable(identifier);
}
private BinaryRow partition(int i) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/TableTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/table/TableTestBase.java
index 09b6121b1..4be85e5cb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/TableTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/TableTestBase.java
@@ -185,8 +185,8 @@ public abstract class TableTestBase {
return messages;
}
- public Table getTableDefault() throws Exception {
- return catalog.getTable(identifier());
+ public FileStoreTable getTableDefault() throws Exception {
+ return (FileStoreTable) catalog.getTable(identifier());
}
private List<CommitMessage> writeOnce(Table table, int time, int size)
throws Exception {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
index 380cdffb8..0c60ea265 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
@@ -23,7 +23,6 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
@@ -135,7 +134,6 @@ public class CdcSinkBuilder<T> {
private DataStreamSink<?> buildForUnawareBucket(DataStream<CdcRecord>
parsed) {
FileStoreTable dataTable = (FileStoreTable) table;
- return new CdcUnawareBucketSink((AppendOnlyFileStoreTable) dataTable,
parallelism)
- .sinkFrom(parsed);
+ return new CdcUnawareBucketSink(dataTable,
parallelism).sinkFrom(parsed);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
index 5c6c1d363..5cd3d6085 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
@@ -21,14 +21,14 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.StoreSinkWrite;
import org.apache.paimon.flink.sink.UnawareBucketSink;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
/** CDC Sink for unaware bucket table. */
public class CdcUnawareBucketSink extends UnawareBucketSink<CdcRecord> {
- public CdcUnawareBucketSink(AppendOnlyFileStoreTable table, Integer
parallelism) {
+ public CdcUnawareBucketSink(FileStoreTable table, Integer parallelism) {
super(table, null, null, parallelism, false);
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index c20aaf612..4985f4829 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -27,7 +27,6 @@ import
org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.Preconditions;
@@ -171,7 +170,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
}
private void buildForUnawareBucket(FileStoreTable table,
DataStream<CdcRecord> parsed) {
- new CdcUnawareBucketSink((AppendOnlyFileStoreTable) table,
parallelism).sinkFrom(parsed);
+ new CdcUnawareBucketSink(table, parallelism).sinkFrom(parsed);
}
private void buildDividedCdcSink() {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 3f197bb8a..313a7a5d4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -23,7 +23,6 @@ import
org.apache.paimon.flink.compact.UnawareBucketCompactionTopoBuilder;
import org.apache.paimon.flink.sink.CompactorSinkBuilder;
import org.apache.paimon.flink.source.CompactorSourceBuilder;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.RuntimeExecutionMode;
@@ -83,8 +82,7 @@ public class CompactAction extends TableActionBase {
switch (fileStoreTable.bucketMode()) {
case UNAWARE:
{
- buildForUnawareBucketCompaction(
- env, (AppendOnlyFileStoreTable) table,
isStreaming);
+ buildForUnawareBucketCompaction(env, fileStoreTable,
isStreaming);
break;
}
case FIXED:
@@ -109,7 +107,7 @@ public class CompactAction extends TableActionBase {
}
private void buildForUnawareBucketCompaction(
- StreamExecutionEnvironment env, AppendOnlyFileStoreTable table,
boolean isStreaming) {
+ StreamExecutionEnvironment env, FileStoreTable table, boolean
isStreaming) {
UnawareBucketCompactionTopoBuilder unawareBucketCompactionTopoBuilder =
new UnawareBucketCompactionTopoBuilder(env,
identifier.getFullName(), table);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index 9fe095888..cffa94dfc 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -30,7 +30,6 @@ import org.apache.paimon.flink.source.CompactorSourceBuilder;
import org.apache.paimon.flink.source.MultiTablesCompactorSourceBuilder;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.options.Options;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.utils.Preconditions;
@@ -66,7 +65,7 @@ public class CompactDatabaseAction extends ActionBase {
private MultiTablesSinkMode databaseCompactMode =
MultiTablesSinkMode.DIVIDED;
- private final Map<String, Table> tableMap = new HashMap<>();
+ private final Map<String, FileStoreTable> tableMap = new HashMap<>();
private Options tableOptions = new Options();
@@ -143,11 +142,12 @@ public class CompactDatabaseAction extends ActionBase {
table.getClass().getName()));
continue;
}
- table =
- table.copy(
- Collections.singletonMap(
-
CoreOptions.WRITE_ONLY.key(), "false"));
- tableMap.put(fullTableName, table);
+ FileStoreTable fileStoreTable =
+ (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.WRITE_ONLY.key(), "false"));
+ tableMap.put(fullTableName, fileStoreTable);
} else {
LOG.debug("The table {} is excluded.",
fullTableName);
}
@@ -165,16 +165,13 @@ public class CompactDatabaseAction extends ActionBase {
ReadableConfig conf =
StreamExecutionEnvironmentUtils.getConfiguration(env);
boolean isStreaming =
conf.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.STREAMING;
- for (Map.Entry<String, Table> entry : tableMap.entrySet()) {
- FileStoreTable fileStoreTable = (FileStoreTable) entry.getValue();
+ for (Map.Entry<String, FileStoreTable> entry : tableMap.entrySet()) {
+ FileStoreTable fileStoreTable = entry.getValue();
switch (fileStoreTable.bucketMode()) {
case UNAWARE:
{
buildForUnawareBucketCompaction(
- env,
- entry.getKey(),
- (AppendOnlyFileStoreTable) entry.getValue(),
- isStreaming);
+ env, entry.getKey(), fileStoreTable,
isStreaming);
break;
}
case FIXED:
@@ -229,7 +226,7 @@ public class CompactDatabaseAction extends ActionBase {
private void buildForUnawareBucketCompaction(
StreamExecutionEnvironment env,
String fullName,
- AppendOnlyFileStoreTable table,
+ FileStoreTable table,
boolean isStreaming) {
UnawareBucketCompactionTopoBuilder unawareBucketCompactionTopoBuilder =
new UnawareBucketCompactionTopoBuilder(env, fullName, table);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
index 84031474a..7d986463d 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
@@ -25,7 +25,7 @@ import
org.apache.paimon.flink.sink.UnawareBucketCompactionSink;
import org.apache.paimon.flink.source.BucketUnawareCompactSource;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -51,14 +51,12 @@ public class UnawareBucketCompactionTopoBuilder {
private final transient StreamExecutionEnvironment env;
private final String tableIdentifier;
- private final AppendOnlyFileStoreTable table;
+ private final FileStoreTable table;
@Nullable private List<Map<String, String>> specifiedPartitions = null;
private boolean isContinuous = false;
public UnawareBucketCompactionTopoBuilder(
- StreamExecutionEnvironment env,
- String tableIdentifier,
- AppendOnlyFileStoreTable table) {
+ StreamExecutionEnvironment env, String tableIdentifier,
FileStoreTable table) {
this.env = env;
this.tableIdentifier = tableIdentifier;
this.table = table;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
index 4ed939fdc..0860ccb3a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
@@ -23,7 +23,7 @@ import org.apache.paimon.append.AppendOnlyCompactionTask;
import org.apache.paimon.flink.source.BucketUnawareCompactSource;
import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
import org.apache.paimon.options.Options;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.utils.ExecutorThreadFactory;
@@ -53,15 +53,14 @@ public class AppendOnlyTableCompactionWorkerOperator
private static final Logger LOG =
LoggerFactory.getLogger(AppendOnlyTableCompactionWorkerOperator.class);
- private final AppendOnlyFileStoreTable table;
+ private final FileStoreTable table;
private final String commitUser;
private transient AppendOnlyFileStoreWrite write;
private transient ExecutorService lazyCompactExecutor;
private transient Queue<Future<CommitMessage>> result;
- public AppendOnlyTableCompactionWorkerOperator(
- AppendOnlyFileStoreTable table, String commitUser) {
+ public AppendOnlyTableCompactionWorkerOperator(FileStoreTable table,
String commitUser) {
super(Options.fromMap(table.options()));
this.table = table;
this.commitUser = commitUser;
@@ -75,7 +74,7 @@ public class AppendOnlyTableCompactionWorkerOperator
@Override
public void open() throws Exception {
LOG.debug("Opened a append-only table compaction worker.");
- this.write = table.store().newWrite(commitUser);
+ this.write = (AppendOnlyFileStoreWrite)
table.store().newWrite(commitUser);
this.result = new LinkedList<>();
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
index 17926093c..b5de897b9 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -20,7 +20,6 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.sink.index.GlobalDynamicBucketSink;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
@@ -142,14 +141,10 @@ public class FlinkSinkBuilder {
private DataStreamSink<?> buildUnawareBucketSink(DataStream<InternalRow>
input) {
checkArgument(
- table instanceof AppendOnlyFileStoreTable,
+ table.primaryKeys().isEmpty(),
"Unaware bucket mode only works with append-only table for
now.");
return new RowUnawareBucketSink(
- (AppendOnlyFileStoreTable) table,
- overwritePartition,
- logSinkFunction,
- parallelism,
- boundedInput)
+ table, overwritePartition, logSinkFunction,
parallelism, boundedInput)
.sinkFrom(input);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
index 3874e2ca7..7434ce07f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
@@ -26,9 +26,7 @@ import org.apache.paimon.flink.PaimonDataStreamSinkProvider;
import org.apache.paimon.flink.log.LogSinkProvider;
import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.options.Options;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -75,11 +73,11 @@ public abstract class FlinkTableSinkBase
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
- if (table instanceof AppendOnlyFileStoreTable) {
+ if (table.primaryKeys().isEmpty()) {
// Don't check this, for example, only inserts are available from
the database, but the
// plan phase contains all changelogs
return requestedMode;
- } else if (table instanceof PrimaryKeyFileStoreTable) {
+ } else {
Options options = Options.fromMap(table.options());
if (options.get(CHANGELOG_PRODUCER) == ChangelogProducer.INPUT) {
return requestedMode;
@@ -101,9 +99,6 @@ public abstract class FlinkTableSinkBase
}
}
return builder.build();
- } else {
- throw new UnsupportedOperationException(
- "Unknown FileStoreTable subclass " +
table.getClass().getName());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
index d838a9aa4..d7b8c76cf 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowUnawareBucketSink.java
@@ -19,7 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -29,7 +29,7 @@ import java.util.Map;
public class RowUnawareBucketSink extends UnawareBucketSink<InternalRow> {
public RowUnawareBucketSink(
- AppendOnlyFileStoreTable table,
+ FileStoreTable table,
Map<String, String> overwritePartitions,
LogSinkFunction logSinkFunction,
Integer parallelism,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
index 4150b7f3a..c45fd168f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
@@ -28,9 +28,7 @@ import org.apache.paimon.predicate.AllPrimaryKeyEqualVisitor;
import org.apache.paimon.predicate.OnlyPartitionKeyEqualVisitor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableUtils;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -93,7 +91,7 @@ public abstract class SupportsRowLevelOperationFlinkTableSink
extends FlinkTable
// unsupported. Similarly, it is not allowed to update the primary key
column when updating
// the column of PrimaryKeyFileStoreTable, because the old data cannot
be handled
// correctly.
- if (table instanceof PrimaryKeyFileStoreTable) {
+ if (table.primaryKeys().size() > 0) {
Options options = Options.fromMap(table.options());
Set<String> primaryKeys = new HashSet<>(table.primaryKeys());
updatedColumns.forEach(
@@ -126,15 +124,10 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
MERGE_ENGINE.key(),
MergeEngine.DEDUPLICATE,
MergeEngine.PARTIAL_UPDATE));
- } else if (table instanceof AppendOnlyFileStoreTable) {
- throw new UnsupportedOperationException(
- String.format(
- "%s can not support update, because there is no
primary key.",
- table.getClass().getName()));
} else {
throw new UnsupportedOperationException(
String.format(
- "%s can not support update, because it is an
unknown subclass of FileStoreTable.",
+ "%s can not support update, because there is no
primary key.",
table.getClass().getName()));
}
}
@@ -184,7 +177,7 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
}
private void validateDeletable() {
- if (table instanceof PrimaryKeyFileStoreTable) {
+ if (table.primaryKeys().size() > 0) {
Options options = Options.fromMap(table.options());
if (options.get(MERGE_ENGINE) == MergeEngine.DEDUPLICATE) {
return;
@@ -193,15 +186,10 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
String.format(
"merge engine '%s' can not support delete,
currently only %s can support delete.",
options.get(MERGE_ENGINE),
MergeEngine.DEDUPLICATE));
- } else if (table instanceof AppendOnlyFileStoreTable) {
- throw new UnsupportedOperationException(
- String.format(
- "table '%s' can not support delete, because there
is no primary key.",
- table.getClass().getName()));
} else {
throw new UnsupportedOperationException(
String.format(
- "%s can not support delete, because it is an
unknown subclass of FileStoreTable.",
+ "table '%s' can not support delete, because there
is no primary key.",
table.getClass().getName()));
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
index 5c027a326..cf825cec7 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketCompactionSink.java
@@ -20,7 +20,7 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.append.AppendOnlyCompactionTask;
import org.apache.paimon.manifest.ManifestCommittable;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
@@ -29,15 +29,15 @@ import
org.apache.flink.streaming.api.operators.OneInputStreamOperator;
/** Compaction Sink for unaware-bucket table. */
public class UnawareBucketCompactionSink extends
FlinkSink<AppendOnlyCompactionTask> {
- private final AppendOnlyFileStoreTable table;
+ private final FileStoreTable table;
- public UnawareBucketCompactionSink(AppendOnlyFileStoreTable table) {
+ public UnawareBucketCompactionSink(FileStoreTable table) {
super(table, true);
this.table = table;
}
public static DataStreamSink<?> sink(
- AppendOnlyFileStoreTable table,
DataStream<AppendOnlyCompactionTask> input) {
+ FileStoreTable table, DataStream<AppendOnlyCompactionTask> input) {
return new UnawareBucketCompactionSink(table).sinkFrom(input);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
index abfb1c845..ed4bca7b2 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketSink.java
@@ -19,7 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.flink.compact.UnawareBucketCompactionTopoBuilder;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ExecutionOptions;
@@ -37,14 +37,14 @@ import java.util.Map;
*/
public abstract class UnawareBucketSink<T> extends FlinkWriteSink<T> {
- protected final AppendOnlyFileStoreTable table;
+ protected final FileStoreTable table;
protected final LogSinkFunction logSinkFunction;
@Nullable protected final Integer parallelism;
protected final boolean boundedInput;
public UnawareBucketSink(
- AppendOnlyFileStoreTable table,
+ FileStoreTable table,
@Nullable Map<String, String> overwritePartitions,
LogSinkFunction logSinkFunction,
@Nullable Integer parallelism,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
index 83cf3600a..e6c301ded 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
@@ -22,7 +22,7 @@ import org.apache.paimon.append.AppendOnlyCompactionTask;
import org.apache.paimon.append.AppendOnlyTableCompactionCoordinator;
import org.apache.paimon.flink.sink.CompactionTaskTypeInfo;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.EndOfScanException;
import org.apache.flink.api.connector.source.Boundedness;
@@ -52,7 +52,7 @@ public class BucketUnawareCompactSource extends
RichSourceFunction<AppendOnlyCom
private static final Logger LOG =
LoggerFactory.getLogger(BucketUnawareCompactSource.class);
private static final String COMPACTION_COORDINATOR_NAME = "Compaction
Coordinator";
- private final AppendOnlyFileStoreTable table;
+ private final FileStoreTable table;
private final boolean streaming;
private final long scanInterval;
private final Predicate filter;
@@ -61,7 +61,7 @@ public class BucketUnawareCompactSource extends
RichSourceFunction<AppendOnlyCom
private volatile boolean isRunning = true;
public BucketUnawareCompactSource(
- AppendOnlyFileStoreTable table,
+ FileStoreTable table,
boolean isStreaming,
long scanInterval,
@Nullable Predicate filter) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
index 208691477..dd3783deb 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -31,8 +31,6 @@ import org.apache.paimon.flink.lookup.FileStoreLookupFunction;
import org.apache.paimon.flink.lookup.LookupRuntimeProviderFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
-import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.utils.Projection;
@@ -145,9 +143,9 @@ public class DataTableSource extends FlinkTableSource {
return ChangelogMode.insertOnly();
}
- if (table instanceof AppendOnlyFileStoreTable) {
+ if (table.primaryKeys().isEmpty()) {
return ChangelogMode.insertOnly();
- } else if (table instanceof PrimaryKeyFileStoreTable) {
+ } else {
Options options = Options.fromMap(table.options());
if (new CoreOptions(options).mergeEngine() ==
CoreOptions.MergeEngine.FIRST_ROW) {
@@ -169,11 +167,6 @@ public class DataTableSource extends FlinkTableSource {
&& options.get(LOG_CHANGELOG_MODE) ==
LogChangelogMode.ALL
? ChangelogMode.all()
: ChangelogMode.upsert();
- } else {
- throw new UnsupportedOperationException(
- "Unsupported Table subclass "
- + table.getClass().getName()
- + " for streaming mode.");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForDynamicBucketITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForDynamicBucketITCase.java
index adc19eb4f..ff3616e65 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForDynamicBucketITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForDynamicBucketITCase.java
@@ -24,11 +24,11 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.KeyValueFileStoreScan;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -41,6 +41,7 @@ import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Random;
@@ -57,23 +58,18 @@ public class SortCompactActionForDynamicBucketITCase
extends ActionITCaseBase {
PredicateBuilder predicateBuilder = new
PredicateBuilder(getTable().rowType());
Predicate predicate = predicateBuilder.between(1, 100L, 200L);
- List<ManifestEntry> files = ((FileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilter =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
zorder(Arrays.asList("f2", "f1"));
- List<ManifestEntry> filesZorder =
- ((FileStoreTable) getTable()).store().newScan().plan().files();
+ List<ManifestEntry> filesZorder =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilterZorder =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
@@ -91,23 +87,18 @@ public class SortCompactActionForDynamicBucketITCase
extends ActionITCaseBase {
// order f2,f1 will make predicate of f1 perform worse.
order(Arrays.asList("f2", "f1"));
- List<ManifestEntry> files = ((FileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilter =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
zorder(Arrays.asList("f2", "f1"));
- List<ManifestEntry> filesZorder =
- ((FileStoreTable) getTable()).store().newScan().plan().files();
+ List<ManifestEntry> filesZorder =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilterZorder =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
@@ -128,23 +119,18 @@ public class SortCompactActionForDynamicBucketITCase
extends ActionITCaseBase {
BinaryString.fromString("000000000" + 100),
BinaryString.fromString("000000000" + 200));
- List<ManifestEntry> files = ((FileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilter =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
- zorder(Arrays.asList("f4"));
+ zorder(Collections.singletonList("f4"));
- List<ManifestEntry> filesZorder =
- ((FileStoreTable) getTable()).store().newScan().plan().files();
+ List<ManifestEntry> filesZorder =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilterZorder =
- ((PrimaryKeyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((KeyValueFileStoreScan) getTable().store().newScan())
.withValueFilter(predicate)
.plan()
.files();
@@ -238,8 +224,8 @@ public class SortCompactActionForDynamicBucketITCase
extends ActionITCaseBase {
catalog.createTable(identifier(), schema(), true);
}
- private Table getTable() throws Exception {
- return catalog.getTable(identifier());
+ private FileStoreTable getTable() throws Exception {
+ return (FileStoreTable) catalog.getTable(identifier());
}
private Identifier identifier() {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
index dc9c56879..cadd70e89 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
@@ -26,10 +26,11 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.operation.AppendOnlyFileStoreScan;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -92,8 +93,7 @@ public class SortCompactActionForUnawareBucketITCase extends
ActionITCaseBase {
Assertions.assertThatCode(() -> order(Arrays.asList("f1", "f2")))
.doesNotThrowAnyException();
- List<ManifestEntry> files =
- ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
ManifestEntry entry = files.get(0);
DataSplit dataSplit =
@@ -118,7 +118,7 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
Assertions.assertThatCode(() -> order(Arrays.asList("f2", "f1")))
.doesNotThrowAnyException();
- files = ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ files = getTable().store().newScan().plan().files();
entry = files.get(0);
dataSplit =
@@ -161,12 +161,9 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
PredicateBuilder predicateBuilder = new
PredicateBuilder(getTable().rowType());
Predicate predicate = predicateBuilder.between(1, 100, 200);
- List<ManifestEntry> files =
- ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilter =
- ((AppendOnlyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((AppendOnlyFileStoreScan) getTable().store().newScan())
.withFilter(predicate)
.plan()
.files();
@@ -175,11 +172,9 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
zorder(Arrays.asList("f2", "f1"));
- files = ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ files = getTable().store().newScan().plan().files();
filesFilter =
- ((AppendOnlyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((AppendOnlyFileStoreScan) getTable().store().newScan())
.withFilter(predicate)
.plan()
.files();
@@ -194,23 +189,17 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
PredicateBuilder predicateBuilder = new
PredicateBuilder(getTable().rowType());
Predicate predicate = predicateBuilder.between(1, 10, 20);
- List<ManifestEntry> filesZorder =
- ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> filesZorder =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilterZorder =
- ((AppendOnlyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((AppendOnlyFileStoreScan) getTable().store().newScan())
.withFilter(predicate)
.plan()
.files();
order(Arrays.asList("f2", "f1"));
- List<ManifestEntry> filesOrder =
- ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> filesOrder =
getTable().store().newScan().plan().files();
List<ManifestEntry> filesFilterOrder =
- ((AppendOnlyFileStoreTable) getTable())
- .store()
- .newScan()
+ ((AppendOnlyFileStoreScan) getTable().store().newScan())
.withFilter(predicate)
.plan()
.files();
@@ -246,15 +235,14 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
prepareSameData(200);
Assertions.assertThatCode(() -> order(Collections.singletonList("f1")))
.doesNotThrowAnyException();
- List<ManifestEntry> files =
- ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ List<ManifestEntry> files =
getTable().store().newScan().plan().files();
Assertions.assertThat(files.size()).isEqualTo(3);
dropTable();
prepareSameData(200);
Assertions.assertThatCode(() -> zorder(Arrays.asList("f1", "f2")))
.doesNotThrowAnyException();
- files = ((AppendOnlyFileStoreTable)
getTable()).store().newScan().plan().files();
+ files = getTable().store().newScan().plan().files();
Assertions.assertThat(files.size()).isEqualTo(3);
}
@@ -370,8 +358,8 @@ public class SortCompactActionForUnawareBucketITCase
extends ActionITCaseBase {
return messages;
}
- private Table getTable() throws Exception {
- return catalog.getTable(identifier());
+ private FileStoreTable getTable() throws Exception {
+ return (FileStoreTable) catalog.getTable(identifier());
}
private static List<CommitMessage> writeOnce(Table table, int p, int size)
throws Exception {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java
index 26c8b8427..ad4a2a313 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java
@@ -27,7 +27,6 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -50,8 +49,7 @@ public class AppendOnlyTableCompactionWorkerOperatorTest
extends TableTestBase {
public void testAsyncCompactionWorks() throws Exception {
createTableDefault();
AppendOnlyTableCompactionWorkerOperator workerOperator =
- new AppendOnlyTableCompactionWorkerOperator(
- (AppendOnlyFileStoreTable) getTableDefault(), "user");
+ new AppendOnlyTableCompactionWorkerOperator(getTableDefault(),
"user");
// write 200 files
List<CommitMessage> commitMessages = writeDataDefault(200, 20);
@@ -103,8 +101,7 @@ public class AppendOnlyTableCompactionWorkerOperatorTest
extends TableTestBase {
public void testAsyncCompactionFileDeletedWhenShutdown() throws Exception {
createTableDefault();
AppendOnlyTableCompactionWorkerOperator workerOperator =
- new AppendOnlyTableCompactionWorkerOperator(
- (AppendOnlyFileStoreTable) getTableDefault(), "user");
+ new AppendOnlyTableCompactionWorkerOperator(getTableDefault(),
"user");
// write 200 files
List<CommitMessage> commitMessages = writeDataDefault(200, 40);
@@ -125,7 +122,7 @@ public class AppendOnlyTableCompactionWorkerOperatorTest
extends TableTestBase {
LocalFileIO localFileIO = LocalFileIO.create();
DataFilePathFactory dataFilePathFactory =
- ((AppendOnlyFileStoreTable) getTableDefault())
+ getTableDefault()
.store()
.pathFactory()
.createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
index 6e32b61c2..6b6ae8c3a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/statistics/AppendOnlyTableStatisticsTest.java
@@ -23,11 +23,10 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
-/** Statistics tests for {@link AppendOnlyFileStoreTable}. */
+/** Statistics tests for append only tables. */
public class AppendOnlyTableStatisticsTest extends
FileStoreTableStatisticsTestBase {
@Override
protected FileStoreTable createStoreTable() throws Exception {
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 745a4edc2..91a607aaf 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
@@ -27,15 +27,14 @@ import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
-import org.apache.paimon.table.PrimaryKeyFileStoreTable;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
-/** Statistics tests for {@link PrimaryKeyFileStoreTable}. */
+/** Statistics tests for table with primary keys. */
public class PrimaryKeyTableStatisticsTest extends
FileStoreTableStatisticsTestBase {
- /** {@link PrimaryKeyFileStoreTable} does not support filter value. */
+ /** Primary key tables does not support filter value. */
@Test
public void testTableFilterValueStatistics() throws Exception {
FileStoreTable table = writeData();
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
index c199d7e89..642236cfc 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
@@ -30,7 +30,6 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.migrate.FileMetaUtils;
import org.apache.paimon.migrate.Migrator;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -187,7 +186,7 @@ public class HiveMigrator implements Migrator {
}
private void checkPaimonTable(FileStoreTable paimonTable) {
- if (!(paimonTable instanceof AppendOnlyFileStoreTable)) {
+ if (paimonTable.primaryKeys().size() > 0) {
throw new IllegalArgumentException(
"Hive migrator only support append only table target
table");
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 13c086e24..4d5cc934d 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -32,7 +32,6 @@ import org.apache.paimon.spark.DynamicOverWrite$;
import org.apache.paimon.spark.SparkUtils;
import org.apache.paimon.spark.commands.WriteIntoPaimonTable;
import org.apache.paimon.spark.sort.TableSorter;
-import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchTableCommit;
@@ -246,9 +245,8 @@ public class CompactProcedure extends BaseProcedure {
private void compactUnAwareBucketTable(
FileStoreTable table, @Nullable Predicate filter, JavaSparkContext
javaSparkContext) {
- AppendOnlyFileStoreTable fileStoreTable = (AppendOnlyFileStoreTable)
table;
List<AppendOnlyCompactionTask> compactionTasks =
- new AppendOnlyTableCompactionCoordinator(fileStoreTable,
false, filter).run();
+ new AppendOnlyTableCompactionCoordinator(table, false,
filter).run();
if (compactionTasks.isEmpty()) {
return;
}
@@ -271,7 +269,8 @@ public class CompactProcedure extends BaseProcedure {
(FlatMapFunction<Iterator<byte[]>,
CommitMessage>)
taskIterator -> {
AppendOnlyFileStoreWrite write =
-
fileStoreTable.store().newWrite(commitUser);
+ (AppendOnlyFileStoreWrite)
+
table.store().newWrite(commitUser);
CompactionTaskSerializer ser =
new
CompactionTaskSerializer();
ArrayList<CommitMessage> messages
= new ArrayList<>();
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index 64f22d91f..754f2de8b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -17,8 +17,7 @@
*/
package org.apache.paimon.spark
-import org.apache.paimon.table.{AppendOnlyFileStoreTable, Table}
-import org.apache.paimon.table.source.ReadBuilder
+import org.apache.paimon.table.Table
import org.apache.spark.sql.connector.read.SupportsPushDownLimit
@@ -27,7 +26,7 @@ class PaimonScanBuilder(table: Table)
with SupportsPushDownLimit {
override def pushLimit(limit: Int): Boolean = {
- if (table.isInstanceOf[AppendOnlyFileStoreTable]) {
+ if (table.primaryKeys().isEmpty) {
pushDownLimit = Some(limit)
}
// just make a best effort to push down limit
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RowLevelHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RowLevelHelper.scala
index 522d67ccc..63112ecd1 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RowLevelHelper.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/RowLevelHelper.scala
@@ -20,26 +20,21 @@ package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.CoreOptions.MERGE_ENGINE
import org.apache.paimon.options.Options
import org.apache.paimon.spark.SparkTable
-import org.apache.paimon.table.{FileStoreTable, PrimaryKeyFileStoreTable,
Table}
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, AttributeSet, BinaryExpression, EqualTo, Expression,
SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical.Assignment
-import scala.collection.JavaConverters._
-
trait RowLevelHelper extends SQLConfHelper {
val operation: RowLevelOp
protected def checkPaimonTable(table: SparkTable): Unit = {
- val paimonTable = table.getTable match {
- case pkTable: PrimaryKeyFileStoreTable => pkTable
- case _: FileStoreTable =>
- throw new UnsupportedOperationException(
- s"Only support to $operation table with primary keys.")
- case _ =>
- throw new UnsupportedOperationException(s"Can't $operation a non-file
store table.")
+ val paimonTable = if (table.getTable.primaryKeys().size() > 0) {
+ table.getTable
+ } else {
+ throw new UnsupportedOperationException(
+ s"Only support to $operation table with primary keys.")
}
val options = Options.fromMap(paimonTable.options)
