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 494578140 [core] Make AbstractFileStoreTable package private (#2779)
494578140 is described below
commit 49457814002366680afcceb106ae8ccb76aa7b37
Author: tsreaper <[email protected]>
AuthorDate: Wed Jan 24 14:06:09 2024 +0800
[core] Make AbstractFileStoreTable package private (#2779)
---
.../org/apache/paimon/catalog/AbstractCatalog.java | 7 ++-----
.../paimon/crosspartition/GlobalIndexAssigner.java | 10 +++++-----
.../org/apache/paimon/migrate/FileMetaUtils.java | 10 ++++------
.../paimon/table/AbstractFileStoreTable.java | 10 ++++------
.../paimon/table/AppendOnlyFileStoreTable.java | 4 ++--
.../org/apache/paimon/table/FileStoreTable.java | 2 ++
.../paimon/table/PrimaryKeyFileStoreTable.java | 4 ++--
.../paimon/table/system/ReadOptimizedTable.java | 5 ++---
.../paimon/catalog/PrimaryKeyTableTestBase.java | 6 +++---
.../paimon/operation/PartitionExpireTest.java | 3 +--
.../org/apache/paimon/stats/StatsTableTest.java | 4 ++--
.../paimon/table/IndexFileExpireTableTest.java | 9 +++++----
.../cdc/kafka/KafkaSyncTableActionITCase.java | 5 ++---
.../cdc/pulsar/PulsarSyncTableActionITCase.java | 5 ++---
.../paimon/flink/action/DropPartitionAction.java | 3 +--
.../apache/paimon/flink/sink/FlinkTableSink.java | 4 ++--
.../flink/sink/HashBucketAssignerOperator.java | 6 +++---
.../SupportsRowLevelOperationFlinkTableSink.java | 4 ++--
.../paimon/flink/DynamicBucketTableITCase.java | 12 +++++++-----
.../apache/paimon/hive/migrate/HiveMigrator.java | 22 ++++++++++++----------
.../paimon/spark/PaimonPartitionManagement.scala | 4 ++--
.../apache/paimon/spark/PaimonSparkTestBase.scala | 6 +++---
.../spark/procedure/CompactProcedureTestBase.scala | 6 +++---
23 files changed, 73 insertions(+), 78 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 22f1b4ac8..89fd52f6c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -30,7 +30,6 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
@@ -127,7 +126,7 @@ public abstract class AbstractCatalog implements Catalog {
public void dropPartition(Identifier identifier, Map<String, String>
partitionSpec)
throws TableNotExistException {
Table table = getTable(identifier);
- AbstractFileStoreTable fileStoreTable = (AbstractFileStoreTable) table;
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
FileStoreCommit commit =
fileStoreTable.store().newCommit(UUID.randomUUID().toString());
commit.dropPartitions(
Collections.singletonList(partitionSpec),
BatchWriteBuilder.COMMIT_IDENTIFIER);
@@ -446,8 +445,6 @@ public abstract class AbstractCatalog implements Catalog {
} else if (change instanceof SchemaChange.RenameColumn) {
SchemaChange.RenameColumn rename = (SchemaChange.RenameColumn)
change;
fieldNames.add(rename.newName());
- } else {
- // do nothing
}
}
validateFieldNameCaseInsensitive(fieldNames);
@@ -459,7 +456,7 @@ public abstract class AbstractCatalog implements Catalog {
private void validateAutoCreateClose(Map<String, String> options) {
checkArgument(
- !Boolean.valueOf(
+ !Boolean.parseBoolean(
options.getOrDefault(
CoreOptions.AUTO_CREATE.key(),
CoreOptions.AUTO_CREATE.defaultValue().toString())),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index f003ffc10..5b4ee8a43 100644
---
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -37,7 +37,7 @@ import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.PartitionKeyExtractor;
import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
@@ -78,12 +78,11 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
private static final String INDEX_NAME = "keyIndex";
- private final AbstractFileStoreTable table;
+ private final FileStoreTable table;
private transient IOManager ioManager;
private transient int bucketIndex;
- private transient ProjectToRowFunction setPartition;
private transient boolean bootstrap;
private transient BinaryExternalSortBuffer bootstrapKeys;
private transient RowBuffer bootstrapRecords;
@@ -103,7 +102,7 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
private transient ExistingProcessor existingProcessor;
public GlobalIndexAssigner(Table table) {
- this.table = (AbstractFileStoreTable) table;
+ this.table = (FileStoreTable) table;
}
// ================== Start Public API ===================
@@ -122,7 +121,8 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
RowType bootstrapType = IndexBootstrap.bootstrapType(table.schema());
this.bucketIndex = bootstrapType.getFieldCount() - 1;
- this.setPartition = new ProjectToRowFunction(table.rowType(),
table.partitionKeys());
+ ProjectToRowFunction setPartition =
+ new ProjectToRowFunction(table.rowType(),
table.partitionKeys());
CoreOptions coreOptions = table.coreOptions();
this.targetBucketRowNumber = (int)
coreOptions.dynamicBucketTargetRowNum();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java
b/paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java
index 57b4bf8a2..2b4612229 100644
--- a/paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/migrate/FileMetaUtils.java
@@ -33,7 +33,7 @@ import org.apache.paimon.io.NewFilesIncrement;
import org.apache.paimon.statistics.FieldStatsCollector;
import org.apache.paimon.stats.BinaryTableStats;
import org.apache.paimon.stats.FieldStatsArraySerializer;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -104,14 +104,12 @@ public class FileMetaUtils {
try {
FieldStatsCollector.Factory[] factories =
StatsCollectorFactories.createStatsFactories(
- ((AbstractFileStoreTable) table).coreOptions(),
+ ((FileStoreTable) table).coreOptions(),
table.rowType().getFieldNames());
TableStatsExtractor tableStatsExtractor =
FileFormat.getFileFormat(
- ((AbstractFileStoreTable) table)
- .coreOptions()
- .toConfiguration(),
+ ((FileStoreTable)
table).coreOptions().toConfiguration(),
format)
.createStatsExtractor(table.rowType(), factories)
.orElseThrow(
@@ -167,7 +165,7 @@ public class FileMetaUtils {
stats,
0,
0,
- ((AbstractFileStoreTable) table).schema().id());
+ ((FileStoreTable) table).schema().id());
}
public static BinaryRow writePartitionValue(
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 041470055..9900d9ecf 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
@@ -72,7 +72,7 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Abstract {@link FileStoreTable}. */
-public abstract class AbstractFileStoreTable implements FileStoreTable {
+abstract class AbstractFileStoreTable implements FileStoreTable {
private static final long serialVersionUID = 1L;
@@ -81,7 +81,7 @@ public abstract class AbstractFileStoreTable implements
FileStoreTable {
protected final TableSchema tableSchema;
protected final CatalogEnvironment catalogEnvironment;
- public AbstractFileStoreTable(
+ protected AbstractFileStoreTable(
FileIO fileIO,
Path path,
TableSchema tableSchema,
@@ -165,11 +165,9 @@ public abstract class AbstractFileStoreTable implements
FileStoreTable {
DefaultValueAssigner.create(tableSchema));
}
- public abstract SplitGenerator splitGenerator();
+ protected abstract SplitGenerator splitGenerator();
- public abstract boolean supportStreamingReadOverwrite();
-
- public abstract BiConsumer<FileStoreScan, Predicate>
nonPartitionFilterConsumer();
+ protected abstract BiConsumer<FileStoreScan, Predicate>
nonPartitionFilterConsumer();
protected abstract FileStoreTable copy(TableSchema newTableSchema);
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 0ebcd44ae..678453729 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
@@ -89,7 +89,7 @@ public class AppendOnlyFileStoreTable extends
AbstractFileStoreTable {
}
@Override
- public SplitGenerator splitGenerator() {
+ protected SplitGenerator splitGenerator() {
return new AppendOnlySplitGenerator(
store().options().splitTargetSize(),
store().options().splitOpenFileCost(),
@@ -106,7 +106,7 @@ public class AppendOnlyFileStoreTable extends
AbstractFileStoreTable {
}
@Override
- public BiConsumer<FileStoreScan, Predicate> nonPartitionFilterConsumer() {
+ protected BiConsumer<FileStoreScan, Predicate>
nonPartitionFilterConsumer() {
return (scan, predicate) -> ((AppendOnlyFileStoreScan)
scan).withFilter(predicate);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
index b457f562a..b183d6ad6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FileStoreTable.java
@@ -102,4 +102,6 @@ public interface FileStoreTable extends DataTable {
default BinaryTableStats getSchemaFieldStats(DataFileMeta dataFileMeta) {
return dataFileMeta.valueStats();
}
+
+ boolean supportStreamingReadOverwrite();
}
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 ef91edf31..d815aae9f 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
@@ -117,7 +117,7 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
}
@Override
- public SplitGenerator splitGenerator() {
+ protected SplitGenerator splitGenerator() {
return new MergeTreeSplitGenerator(
store().newKeyComparator(),
store().options().splitTargetSize(),
@@ -130,7 +130,7 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
}
@Override
- public BiConsumer<FileStoreScan, Predicate> nonPartitionFilterConsumer() {
+ protected BiConsumer<FileStoreScan, Predicate>
nonPartitionFilterConsumer() {
return (scan, predicate) -> {
// currently we can only perform filter push down on keys
// consider this case:
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
index 555423734..8daff265f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
@@ -22,7 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.operation.DefaultValueAssigner;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.ReadonlyTable;
@@ -56,10 +55,10 @@ public class ReadOptimizedTable implements DataTable,
ReadonlyTable {
public static final String READ_OPTIMIZED = "ro";
- private final AbstractFileStoreTable dataTable;
+ private final FileStoreTable dataTable;
public ReadOptimizedTable(FileStoreTable dataTable) {
- this.dataTable = (AbstractFileStoreTable) dataTable;
+ this.dataTable = dataTable;
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/PrimaryKeyTableTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/PrimaryKeyTableTestBase.java
index cc765786f..4c334e5bf 100644
---
a/paimon-core/src/test/java/org/apache/paimon/catalog/PrimaryKeyTableTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/PrimaryKeyTableTestBase.java
@@ -22,7 +22,7 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.TraceableFileIO;
@@ -42,7 +42,7 @@ public abstract class PrimaryKeyTableTestBase {
@TempDir protected java.nio.file.Path tempPath;
- protected AbstractFileStoreTable table;
+ protected FileStoreTable table;
protected String commitUser;
@BeforeEach
@@ -63,7 +63,7 @@ public abstract class PrimaryKeyTableTestBase {
.options(tableOptions().toMap())
.build();
catalog.createTable(identifier, schema, true);
- table = (AbstractFileStoreTable) catalog.getTable(identifier);
+ table = (FileStoreTable) catalog.getTable(identifier);
commitUser = UUID.randomUUID().toString();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
index 564d17787..94dc3cfe4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
@@ -26,7 +26,6 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.CommitMessage;
@@ -255,6 +254,6 @@ public class PartitionExpireTest {
options.put(PARTITION_EXPIRATION_TIME.key(), "2 d");
options.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "1
d");
options.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(),
"yyyyMMdd");
- return ((AbstractFileStoreTable)
table.copy(options)).store().newPartitionExpire("");
+ return table.copy(options).store().newPartitionExpire("");
}
}
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 23ff1e822..ce98052e5 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
@@ -27,7 +27,7 @@ import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.types.DataTypes;
@@ -64,7 +64,7 @@ public class StatsTableTest extends TableTestBase {
GenericRow.of(1, 3, 1),
GenericRow.of(2, 1, 1));
- AbstractFileStoreTable storeTable = (AbstractFileStoreTable) table;
+ FileStoreTable storeTable = (FileStoreTable) table;
AbstractFileStore<?> store = (AbstractFileStore<?>) storeTable.store();
String manifestListFile =
storeTable.snapshotManager().latestSnapshot().deltaManifestList();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
index 98816dd04..b7117cf0a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
@@ -106,7 +106,7 @@ public class IndexFileExpireTableTest extends
PrimaryKeyTableTestBase {
assertThat(indexFileSize()).isEqualTo(5);
assertThat(indexManifestSize()).isEqualTo(3);
- TagManager tagManager = new TagManager(LocalFileIO.create(),
table.path);
+ TagManager tagManager = new TagManager(LocalFileIO.create(),
table.location());
checkIndexFiles(tagManager.taggedSnapshot("tag3"));
checkIndexFiles(tagManager.taggedSnapshot("tag5"));
}
@@ -133,7 +133,7 @@ public class IndexFileExpireTableTest extends
PrimaryKeyTableTestBase {
expire.expireUntil(1, 7);
table.deleteTag("tag3");
- TagManager tagManager = new TagManager(LocalFileIO.create(),
table.path);
+ TagManager tagManager = new TagManager(LocalFileIO.create(),
table.location());
checkIndexFiles(7);
checkIndexFiles(tagManager.taggedSnapshot("tag5"));
assertThat(indexFileSize()).isEqualTo(4);
@@ -241,11 +241,12 @@ public class IndexFileExpireTableTest extends
PrimaryKeyTableTestBase {
}
private long indexFileSize() throws IOException {
- return LocalFileIO.create().listStatus(new Path(table.path,
"index")).length;
+ return LocalFileIO.create().listStatus(new Path(table.location(),
"index")).length;
}
private long indexManifestSize() throws IOException {
- return Arrays.stream(LocalFileIO.create().listStatus(new
Path(table.path, "manifest")))
+ return Arrays.stream(
+ LocalFileIO.create().listStatus(new
Path(table.location(), "manifest")))
.filter(s -> s.getPath().getName().startsWith("index-"))
.count();
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
index aabb5d53c..da38b2a0c 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
@@ -22,7 +22,6 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.action.cdc.MessageQueueSchemaUtils;
import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
@@ -652,8 +651,8 @@ public class KafkaSyncTableActionITCase extends
KafkaActionITCaseBase {
syncTableActionBuilder(kafkaConfig).withTableConfig(config).build();
runActionWithDefaultEnv(action);
- AbstractFileStoreTable table =
- (AbstractFileStoreTable) catalog.getTable(new
Identifier(database, tableName));
+ FileStoreTable table =
+ (FileStoreTable) catalog.getTable(new Identifier(database,
tableName));
while (true) {
if (table.snapshotManager().snapshotCount() > 0
&& table.snapshotManager().latestSnapshot().watermark()
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
index 43d396008..a4fb332d5 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
@@ -19,7 +19,6 @@
package org.apache.paimon.flink.action.cdc.pulsar;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -220,8 +219,8 @@ public class PulsarSyncTableActionITCase extends
PulsarActionITCaseBase {
.build();
runActionWithDefaultEnv(action);
- AbstractFileStoreTable table =
- (AbstractFileStoreTable) catalog.getTable(new
Identifier(database, tableName));
+ FileStoreTable table =
+ (FileStoreTable) catalog.getTable(new Identifier(database,
tableName));
while (true) {
if (table.snapshotManager().snapshotCount() > 0
&& table.snapshotManager().latestSnapshot().watermark()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
index a1bd77544..d6a70f375 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DropPartitionAction.java
@@ -19,7 +19,6 @@
package org.apache.paimon.flink.action;
import org.apache.paimon.operation.FileStoreCommit;
-import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -49,7 +48,7 @@ public class DropPartitionAction extends TableActionBase {
this.partitions = partitions;
- AbstractFileStoreTable fileStoreTable = (AbstractFileStoreTable) table;
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
this.commit =
fileStoreTable.store().newCommit(UUID.randomUUID().toString());
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
index 98bb6a69e..50bc45b75 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
@@ -20,7 +20,7 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.operation.FileStoreCommit;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -47,7 +47,7 @@ public class FlinkTableSink extends
SupportsRowLevelOperationFlinkTableSink
@Override
public void executeTruncation() {
FileStoreCommit commit =
- ((AbstractFileStoreTable)
table).store().newCommit(UUID.randomUUID().toString());
+ ((FileStoreTable)
table).store().newCommit(UUID.randomUUID().toString());
long identifier = BatchWriteBuilder.COMMIT_IDENTIFIER;
commit.purgeTable(identifier);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/HashBucketAssignerOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/HashBucketAssignerOperator.java
index b0b9cf2b9..9713421fd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/HashBucketAssignerOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/HashBucketAssignerOperator.java
@@ -22,7 +22,7 @@ import org.apache.paimon.index.BucketAssigner;
import org.apache.paimon.index.HashBucketAssigner;
import org.apache.paimon.index.SimpleHashBucketAssigner;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.PartitionKeyExtractor;
import org.apache.paimon.utils.MathUtils;
@@ -42,7 +42,7 @@ public class HashBucketAssignerOperator<T> extends
AbstractStreamOperator<Tuple2
private final String initialCommitUser;
- private final AbstractFileStoreTable table;
+ private final FileStoreTable table;
private final Integer numAssigners;
private final SerializableFunction<TableSchema, PartitionKeyExtractor<T>>
extractorFunction;
private final boolean overwrite;
@@ -57,7 +57,7 @@ public class HashBucketAssignerOperator<T> extends
AbstractStreamOperator<Tuple2
SerializableFunction<TableSchema, PartitionKeyExtractor<T>>
extractorFunction,
boolean overwrite) {
this.initialCommitUser = commitUser;
- this.table = (AbstractFileStoreTable) table;
+ this.table = (FileStoreTable) table;
this.numAssigners = numAssigners;
this.extractorFunction = extractorFunction;
this.overwrite = overwrite;
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 b628b1cbc..4150b7f3a 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,8 +28,8 @@ 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.AbstractFileStoreTable;
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;
@@ -169,7 +169,7 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
@Override
public Optional<Long> executeDeletion() {
FileStoreCommit commit =
- ((AbstractFileStoreTable)
table).store().newCommit(UUID.randomUUID().toString());
+ ((FileStoreTable)
table).store().newCommit(UUID.randomUUID().toString());
long identifier = BatchWriteBuilder.COMMIT_IDENTIFIER;
if (deletePredicate == null) {
commit.purgeTable(identifier);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
index 06aaa23ca..c4f928ef4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DynamicBucketTableITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink;
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
@@ -25,7 +26,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.manifest.IndexManifestEntry;
-import org.apache.paimon.table.AbstractFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.types.Row;
import org.assertj.core.api.Assertions;
@@ -111,10 +112,9 @@ public class DynamicBucketTableITCase extends
CatalogITCaseBase {
// overwrite the whole table, we should update the index file by this
sql
sql("INSERT OVERWRITE T SELECT * FROM T LIMIT 4");
- AbstractFileStoreTable table =
- (AbstractFileStoreTable)
-
(CatalogFactory.createCatalog(CatalogContext.create(new Path(path))))
- .getTable(Identifier.create("default", "T"));
+ Catalog catalog =
CatalogFactory.createCatalog(CatalogContext.create(new Path(path)));
+
+ FileStoreTable table = (FileStoreTable)
catalog.getTable(Identifier.create("default", "T"));
IndexFileHandler indexFileHandler =
table.store().newIndexFileHandler();
List<BinaryRow> partitions = table.newScan().listPartitions();
List<IndexManifestEntry> entries = new ArrayList<>();
@@ -123,5 +123,7 @@ public class DynamicBucketTableITCase extends
CatalogITCaseBase {
Long records =
entries.stream().map(entry ->
entry.indexFile().rowCount()).reduce(Long::sum).get();
Assertions.assertThat(records).isEqualTo(4);
+
+ catalog.close();
}
}
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 c60c5d3ef..c199d7e89 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,9 +30,10 @@ 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.AbstractFileStoreTable;
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;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
@@ -111,8 +112,7 @@ public class HiveMigrator implements Migrator {
}
try {
- AbstractFileStoreTable paimonTable =
- (AbstractFileStoreTable) hiveCatalog.getTable(identifier);
+ FileStoreTable paimonTable = (FileStoreTable)
hiveCatalog.getTable(identifier);
checkPaimonTable(paimonTable);
List<String> partitionsNames =
@@ -165,7 +165,9 @@ public class HiveMigrator implements Migrator {
throw new RuntimeException("Migrating failed because exception
happens", e);
}
- paimonTable.newBatchWriteBuilder().newCommit().commit(new
ArrayList<>(commitMessages));
+ try (BatchTableCommit commit =
paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.commit(new ArrayList<>(commitMessages));
+ }
} catch (Exception e) {
if (!alreadyExist) {
hiveCatalog.dropTable(identifier, true);
@@ -184,7 +186,7 @@ public class HiveMigrator implements Migrator {
}
}
- private void checkPaimonTable(AbstractFileStoreTable paimonTable) {
+ private void checkPaimonTable(FileStoreTable paimonTable) {
if (!(paimonTable instanceof AppendOnlyFileStoreTable)) {
throw new IllegalArgumentException(
"Hive migrator only support append only table target
table");
@@ -231,7 +233,7 @@ public class HiveMigrator implements Migrator {
FileIO fileIO,
List<String> partitionNames,
Table sourceTable,
- AbstractFileStoreTable paimonTable,
+ FileStoreTable paimonTable,
Map<Path, Path> rollback)
throws Exception {
List<MigrateTask> migrateTasks = new ArrayList<>();
@@ -265,7 +267,7 @@ public class HiveMigrator implements Migrator {
public MigrateTask importUnPartitionedTableTask(
FileIO fileIO,
Table sourceTable,
- AbstractFileStoreTable paimonTable,
+ FileStoreTable paimonTable,
Map<Path, Path> rollback) {
String format =
parseFormat(sourceTable.getSd().getSerdeInfo().toString());
String location = sourceTable.getSd().getLocation();
@@ -274,7 +276,7 @@ public class HiveMigrator implements Migrator {
fileIO, format, location, paimonTable, BinaryRow.EMPTY_ROW,
path, rollback);
}
- private void checkCompatible(Table sourceHiveTable, AbstractFileStoreTable
paimonTable) {
+ private void checkCompatible(Table sourceHiveTable, FileStoreTable
paimonTable) {
List<FieldSchema> sourceFields = new
ArrayList<>(sourceHiveTable.getPartitionKeys());
List<DataField> targetFields =
new ArrayList<>(
@@ -321,7 +323,7 @@ public class HiveMigrator implements Migrator {
private final FileIO fileIO;
private final String format;
private final String location;
- private final AbstractFileStoreTable paimonTable;
+ private final FileStoreTable paimonTable;
private final BinaryRow partitionRow;
private final Path newDir;
private final Map<Path, Path> rollback;
@@ -330,7 +332,7 @@ public class HiveMigrator implements Migrator {
FileIO fileIO,
String format,
String location,
- AbstractFileStoreTable paimonTable,
+ FileStoreTable paimonTable,
BinaryRow partitionRow,
Path newDir,
Map<Path, Path> rollback) {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
index c1872804f..c4149fa91 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
@@ -18,7 +18,7 @@
package org.apache.paimon.spark
import org.apache.paimon.operation.FileStoreCommit
-import org.apache.paimon.table.AbstractFileStoreTable
+import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.BatchWriteBuilder
import org.apache.paimon.types.RowType
import org.apache.paimon.utils.{FileStorePathFactory, RowDataPartitionComputer}
@@ -58,7 +58,7 @@ trait PaimonPartitionManagement extends
SupportsPartitionManagement {
partitionKeys.asScala.toArray)
val partitionMap = rowDataPartitionComputer.generatePartValues(new
SparkRow(tableRowType, row))
getTable match {
- case table: AbstractFileStoreTable =>
+ case table: FileStoreTable =>
val commit: FileStoreCommit =
table.store.newCommit(UUID.randomUUID.toString)
commit.dropPartitions(
Collections.singletonList(partitionMap),
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
index 2aa259ae7..39fb60945 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.options.Options
import org.apache.paimon.spark.catalog.Catalogs
import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
import org.apache.paimon.spark.sql.WithTableOptions
-import org.apache.paimon.table.AbstractFileStoreTable
+import org.apache.paimon.table.FileStoreTable
import org.apache.spark.SparkConf
import org.apache.spark.paimon.Utils
@@ -97,8 +97,8 @@ class PaimonSparkTestBase extends QueryTest with
SharedSparkSession with WithTab
CatalogFactory.createCatalog(catalogContext)
}
- def loadTable(tableName: String): AbstractFileStoreTable = {
- catalog.getTable(Identifier.create(dbName0,
tableName)).asInstanceOf[AbstractFileStoreTable]
+ def loadTable(tableName: String): FileStoreTable = {
+ catalog.getTable(Identifier.create(dbName0,
tableName)).asInstanceOf[FileStoreTable]
}
protected def createRelationV2(tableName: String): LogicalPlan = {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 2a0cc043d..23dee044b 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -19,7 +19,7 @@ package org.apache.paimon.spark.procedure
import org.apache.paimon.Snapshot.CommitKind
import org.apache.paimon.spark.PaimonSparkTestBase
-import org.apache.paimon.table.AbstractFileStoreTable
+import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.execution.streaming.MemoryStream
@@ -422,11 +422,11 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
Assertions.assertThat(where).isEqualTo(whereExpected)
}
- def lastSnapshotCommand(table: AbstractFileStoreTable): CommitKind = {
+ def lastSnapshotCommand(table: FileStoreTable): CommitKind = {
table.snapshotManager().latestSnapshot().commitKind()
}
- def lastSnapshotId(table: AbstractFileStoreTable): Long = {
+ def lastSnapshotId(table: FileStoreTable): Long = {
table.snapshotManager().latestSnapshotId()
}
}
