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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6533329ef [core] Support custom data file name prefix (#4041)
6533329ef is described below
commit 6533329ef91b640c2c1be77ef37b42e09621fdb5
Author: askwang <[email protected]>
AuthorDate: Wed Sep 11 14:32:58 2024 +0800
[core] Support custom data file name prefix (#4041)
---
.../shortcodes/generated/core_configuration.html | 12 +++
.../main/java/org/apache/paimon/CoreOptions.java | 20 ++++
.../java/org/apache/paimon/AbstractFileStore.java | 4 +-
.../java/org/apache/paimon/KeyValueFileStore.java | 4 +-
.../org/apache/paimon/io/DataFilePathFactory.java | 27 ++++--
.../paimon/operation/MergeFileSplitRead.java | 4 +-
.../apache/paimon/utils/FileStorePathFactory.java | 17 +++-
.../apache/paimon/append/AppendOnlyWriterTest.java | 4 +-
.../apache/paimon/format/FileFormatSuffixTest.java | 6 +-
.../apache/paimon/io/DataFilePathFactoryTest.java | 8 +-
.../paimon/io/KeyValueFileReadWriteTest.java | 8 +-
.../apache/paimon/io/RollingFileWriterTest.java | 5 +-
.../paimon/manifest/ManifestFileMetaTestBase.java | 4 +-
.../apache/paimon/manifest/ManifestFileTest.java | 4 +-
.../apache/paimon/manifest/ManifestListTest.java | 4 +-
.../mergetree/ChangelogMergeTreeRewriterTest.java | 8 +-
.../paimon/operation/OrphanFilesCleanTest.java | 9 +-
.../paimon/utils/FileStorePathFactoryTest.java | 8 +-
.../flink/source/TestChangelogDataReadWrite.java | 4 +-
.../apache/paimon/spark/SparkFileIndexITCase.java | 6 +-
.../org/apache/paimon/spark/SparkWriteITCase.java | 107 ++++++++++++++++++++-
.../paimon/spark/SparkWriteWithKyroITCase.java | 2 +-
22 files changed, 239 insertions(+), 36 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 85cdd416d..65fa26a51 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -62,6 +62,12 @@ under the License.
<td>MemorySize</td>
<td>Memory page size for caching.</td>
</tr>
+ <tr>
+ <td><h5>changelog-file.prefix</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify the file name prefix of changelog files.</td>
+ </tr>
<tr>
<td><h5>changelog-producer</h5></td>
<td style="word-wrap: break-word;">none</td>
@@ -320,6 +326,12 @@ under the License.
<td>Map</td>
<td>Define different file format for different level, you can add
the conf like this: 'file.format.per.level' = '0:avro,3:parquet', if the file
format for level is not provided, the default format which set by `file.format`
will be used.</td>
</tr>
+ <tr>
+ <td><h5>data-file.prefix</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify the file name prefix of data files.</td>
+ </tr>
<tr>
<td><h5>force-lookup</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 5df8f0235..a7a744197 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -171,6 +171,18 @@ public class CoreOptions implements Serializable {
.withDescription(
"Default file compression zstd level. For higher
compression rates, it can be configured to 9, but the read and write speed will
significantly decrease.");
+ public static final ConfigOption<String> DATA_FILE_PREFIX =
+ key("data-file.prefix")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Specify the file name prefix of data
files.");
+
+ public static final ConfigOption<String> CHANGELOG_FILE_PREFIX =
+ key("changelog-file.prefix")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Specify the file name prefix of
changelog files.");
+
public static final ConfigOption<MemorySize> FILE_BLOCK_SIZE =
key("file.block-size")
.memoryType()
@@ -1468,6 +1480,14 @@ public class CoreOptions implements Serializable {
return fileFormat.toLowerCase();
}
+ public String dataFilePrefix() {
+ return options.get(DATA_FILE_PREFIX);
+ }
+
+ public String changelogFilePrefix() {
+ return options.get(CHANGELOG_FILE_PREFIX);
+ }
+
public String fieldsDefaultFunc() {
return options.get(FIELDS_DEFAULT_AGG_FUNC);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index c12d95e5c..be4b504ec 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -102,7 +102,9 @@ abstract class AbstractFileStore<T> implements FileStore<T>
{
options.path(),
partitionType,
options.partitionDefaultName(),
- options.fileFormat().getFormatIdentifier());
+ options.fileFormat().getFormatIdentifier(),
+ options.dataFilePrefix(),
+ options.changelogFilePrefix());
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 88fdd01ea..1e950d747 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -203,7 +203,9 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
options.path(),
partitionType,
options.partitionDefaultName(),
- format)));
+ format,
+ options.dataFilePrefix(),
+ options.changelogFilePrefix())));
return pathFactoryMap;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
index 6c7090f3d..a9006eb99 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFilePathFactory.java
@@ -20,6 +20,7 @@ package org.apache.paimon.io;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.StringUtils;
import javax.annotation.concurrent.ThreadSafe;
@@ -30,9 +31,9 @@ import java.util.concurrent.atomic.AtomicInteger;
@ThreadSafe
public class DataFilePathFactory {
- public static final String DATA_FILE_PREFIX = "data-";
+ public static final String DEFAULT_DATA_FILE_PREFIX = "data-";
- public static final String CHANGELOG_FILE_PREFIX = "changelog-";
+ public static final String DEFAULT_CHANGELOG_FILE_PREFIX = "changelog-";
public static final String INDEX_PATH_SUFFIX = ".index";
@@ -41,21 +42,35 @@ public class DataFilePathFactory {
private final AtomicInteger pathCount;
private final String formatIdentifier;
-
- public DataFilePathFactory(Path parent, String formatIdentifier) {
+ private final String dataFilePrefix;
+ private final String changelogFilePrefix;
+
+ public DataFilePathFactory(
+ Path parent,
+ String formatIdentifier,
+ String dataFilePrefix,
+ String changelogFilePrefix) {
this.parent = parent;
this.uuid = UUID.randomUUID().toString();
this.pathCount = new AtomicInteger(0);
this.formatIdentifier = formatIdentifier;
+ this.dataFilePrefix = dataFilePrefix;
+ this.changelogFilePrefix = changelogFilePrefix;
}
public Path newPath() {
- return newPath(DATA_FILE_PREFIX);
+ if (!StringUtils.isBlank(dataFilePrefix)) {
+ return newPath(dataFilePrefix);
+ }
+ return newPath(DEFAULT_DATA_FILE_PREFIX);
}
public Path newChangelogPath() {
- return newPath(CHANGELOG_FILE_PREFIX);
+ if (!StringUtils.isBlank(changelogFilePrefix)) {
+ return newPath(changelogFilePrefix);
+ }
+ return newPath(DEFAULT_CHANGELOG_FILE_PREFIX);
}
private Path newPath(String prefix) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
index 20bee30c1..b30789c43 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
@@ -60,7 +60,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.apache.paimon.io.DataFilePathFactory.CHANGELOG_FILE_PREFIX;
+import static
org.apache.paimon.io.DataFilePathFactory.DEFAULT_CHANGELOG_FILE_PREFIX;
import static org.apache.paimon.predicate.PredicateBuilder.containsFields;
import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
@@ -314,7 +314,7 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
private Optional<String> changelogFile(DataFileMeta fileMeta) {
for (String file : fileMeta.extraFiles()) {
- if (file.startsWith(CHANGELOG_FILE_PREFIX)) {
+ if (file.startsWith(DEFAULT_CHANGELOG_FILE_PREFIX)) {
return Optional.of(file);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
b/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
index a49e6dbc4..df442383a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java
@@ -41,6 +41,8 @@ public class FileStorePathFactory {
private final String uuid;
private final InternalRowPartitionComputer partitionComputer;
private final String formatIdentifier;
+ private final String dataFilePrefix;
+ private final String changelogFilePrefix;
private final AtomicInteger manifestFileCount;
private final AtomicInteger manifestListCount;
@@ -49,12 +51,19 @@ public class FileStorePathFactory {
private final AtomicInteger statsFileCount;
public FileStorePathFactory(
- Path root, RowType partitionType, String defaultPartValue, String
formatIdentifier) {
+ Path root,
+ RowType partitionType,
+ String defaultPartValue,
+ String formatIdentifier,
+ String dataFilePrefix,
+ String changelogFilePrefix) {
this.root = root;
this.uuid = UUID.randomUUID().toString();
this.partitionComputer = getPartitionComputer(partitionType,
defaultPartValue);
this.formatIdentifier = formatIdentifier;
+ this.dataFilePrefix = dataFilePrefix;
+ this.changelogFilePrefix = changelogFilePrefix;
this.manifestFileCount = new AtomicInteger(0);
this.manifestListCount = new AtomicInteger(0);
@@ -97,7 +106,11 @@ public class FileStorePathFactory {
}
public DataFilePathFactory createDataFilePathFactory(BinaryRow partition,
int bucket) {
- return new DataFilePathFactory(bucketPath(partition, bucket),
formatIdentifier);
+ return new DataFilePathFactory(
+ bucketPath(partition, bucket),
+ formatIdentifier,
+ dataFilePrefix,
+ changelogFilePrefix);
}
public Path bucketPath(BinaryRow partition, int bucket) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
index 436f8f584..a4ba849ad 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java
@@ -520,7 +520,9 @@ public class AppendOnlyWriterTest {
private DataFilePathFactory createPathFactory() {
return new DataFilePathFactory(
new Path(tempDir + "/dt=" + PART + "/bucket-0"),
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
}
private AppendOnlyWriter createEmptyWriter(long targetFileSize) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
b/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
index 539ff4757..4aee88298 100644
---
a/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/format/FileFormatSuffixTest.java
@@ -66,7 +66,11 @@ public class FileFormatSuffixTest extends
KeyValueFileReadWriteTest {
assertThat(path.toString().endsWith(format)).isTrue();
DataFilePathFactory dataFilePathFactory =
- new DataFilePathFactory(new Path(tempDir + "/dt=1/bucket-1"),
format);
+ new DataFilePathFactory(
+ new Path(tempDir + "/dt=1/bucket-1"),
+ format,
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
FileFormat fileFormat = FileFormat.fromIdentifier(format, new
Options());
LinkedList<DataFileMeta> toCompact = new LinkedList<>();
CoreOptions options = new CoreOptions(new HashMap<>());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java
index bf741e026..609566258 100644
---
a/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/io/DataFilePathFactoryTest.java
@@ -36,7 +36,9 @@ public class DataFilePathFactoryTest {
DataFilePathFactory pathFactory =
new DataFilePathFactory(
new Path(tempDir + "/bucket-123"),
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
String uuid = pathFactory.uuid();
for (int i = 0; i < 20; i++) {
@@ -60,7 +62,9 @@ public class DataFilePathFactoryTest {
DataFilePathFactory pathFactory =
new DataFilePathFactory(
new Path(tempDir + "/dt=20211224/bucket-123"),
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
String uuid = pathFactory.uuid();
for (int i = 0; i < 20; i++) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
index 9f1405680..3b044d24f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java
@@ -248,7 +248,9 @@ public class KeyValueFileReadWriteTest {
path,
RowType.of(),
CoreOptions.PARTITION_DEFAULT_NAME.defaultValue(),
- format);
+ format,
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
int suggestedFileSize = ThreadLocalRandom.current().nextInt(8192) +
1024;
FileIO fileIO = FileIOFinder.find(path);
Options options = new Options();
@@ -262,7 +264,9 @@ public class KeyValueFileReadWriteTest {
path,
RowType.of(),
CoreOptions.PARTITION_DEFAULT_NAME.defaultValue(),
- CoreOptions.FILE_FORMAT.defaultValue().toString()));
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue()));
return KeyValueFileWriterFactory.builder(
fileIO,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
index d3039e00c..df9f2ac0b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
@@ -73,7 +73,10 @@ public class RollingFileWriterTest {
new Path(tempDir +
"/bucket-0"),
CoreOptions.FILE_FORMAT
.defaultValue()
- .toString())
+ .toString(),
+
CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+
CoreOptions.CHANGELOG_FILE_PREFIX
+
.defaultValue())
.newPath(),
SCHEMA,
fileFormat
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java
index fdb3cb7ff..a937d31a2 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java
@@ -133,7 +133,9 @@ public abstract class ManifestFileMetaTestBase {
path,
getPartitionType(),
"default",
- CoreOptions.FILE_FORMAT.defaultValue()),
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+
CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue()),
Long.MAX_VALUE,
null)
.create();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
index 5ac1eb5f2..a11f3e485 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
@@ -100,7 +100,9 @@ public class ManifestFileTest {
path,
DEFAULT_PART_TYPE,
"default",
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
int suggestedFileSize = ThreadLocalRandom.current().nextInt(8192) +
1024;
FileIO fileIO = FileIOFinder.find(path);
return new ManifestFile.Factory(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java
index 06e106c75..faf1004e1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java
@@ -104,7 +104,9 @@ public class ManifestListTest {
path,
TestKeyValueGenerator.DEFAULT_PART_TYPE,
"default",
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
return new ManifestList.Factory(FileIOFinder.find(path), avro, "zstd",
pathFactory, null)
.create();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
index de9df4a39..59106491d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/ChangelogMergeTreeRewriterTest.java
@@ -137,12 +137,12 @@ public class ChangelogMergeTreeRewriterTest {
.toString()
.startsWith(
DataFilePathFactory
-
.DATA_FILE_PREFIX)
+
.DEFAULT_DATA_FILE_PREFIX)
|| p.getFileName()
.toString()
.startsWith(
DataFilePathFactory
-
.CHANGELOG_FILE_PREFIX))
+
.DEFAULT_CHANGELOG_FILE_PREFIX))
.collect(Collectors.toList());
Assertions.assertEquals(0, files.size());
}
@@ -176,12 +176,12 @@ public class ChangelogMergeTreeRewriterTest {
.toString()
.startsWith(
DataFilePathFactory
-
.DATA_FILE_PREFIX)
+
.DEFAULT_DATA_FILE_PREFIX)
|| p.getFileName()
.toString()
.startsWith(
DataFilePathFactory
-
.CHANGELOG_FILE_PREFIX))
+
.DEFAULT_CHANGELOG_FILE_PREFIX))
.collect(Collectors.toList());
if (rewriteChangelog) {
Assertions.assertEquals(2, files.size()); // changelog + data
file
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
index efb9fb27a..4260cd0e6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/OrphanFilesCleanTest.java
@@ -80,8 +80,8 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import java.util.stream.Collectors;
-import static org.apache.paimon.io.DataFilePathFactory.CHANGELOG_FILE_PREFIX;
-import static org.apache.paimon.io.DataFilePathFactory.DATA_FILE_PREFIX;
+import static
org.apache.paimon.io.DataFilePathFactory.DEFAULT_CHANGELOG_FILE_PREFIX;
+import static
org.apache.paimon.io.DataFilePathFactory.DEFAULT_DATA_FILE_PREFIX;
import static org.apache.paimon.utils.BranchManager.branchPath;
import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX;
import static org.assertj.core.api.Assertions.assertThat;
@@ -584,7 +584,10 @@ public class OrphanFilesCleanTest {
List<Path> corruptedBuckets = randomlyPick(buckets);
for (Path path : corruptedBuckets) {
addNonUsedFiles(
- path, 1, Arrays.asList(DATA_FILE_PREFIX,
CHANGELOG_FILE_PREFIX, "UNKNOWN-"));
+ path,
+ 1,
+ Arrays.asList(
+ DEFAULT_DATA_FILE_PREFIX,
DEFAULT_CHANGELOG_FILE_PREFIX, "UNKNOWN-"));
}
addedFiles += corruptedBuckets.size();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/FileStorePathFactoryTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/FileStorePathFactoryTest.java
index e0031f7c1..994367d0c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/utils/FileStorePathFactoryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/utils/FileStorePathFactoryTest.java
@@ -85,7 +85,9 @@ public class FileStorePathFactoryTest {
new DataType[] {new VarCharType(10), new
IntType()},
new String[] {"dt", "hr"}),
"default",
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
assertPartition("20211224", 16, pathFactory, "/dt=20211224/hr=16");
assertPartition("20211224", null, pathFactory,
"/dt=20211224/hr=default");
@@ -120,6 +122,8 @@ public class FileStorePathFactoryTest {
root,
RowType.builder().build(),
PARTITION_DEFAULT_NAME.defaultValue(),
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
index 66566f873..aab8666fe 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
@@ -105,7 +105,9 @@ public class TestChangelogDataReadWrite {
tablePath,
RowType.of(new IntType()),
"default",
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
this.snapshotManager = new SnapshotManager(LocalFileIO.create(), new
Path(root));
this.commitUser = UUID.randomUUID().toString();
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
index 9b1f62449..7af83186e 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
@@ -72,7 +72,7 @@ public class SparkFileIndexITCase extends SparkWriteITCase {
@BeforeAll
public void startMetastoreAndSpark(@TempDir java.nio.file.Path tempDir) {
- Path warehousePath = new Path("file:///" + tempDir.toString());
+ warehousePath = new Path("file:///" + tempDir.toString());
spark =
SparkSession.builder()
.master("local[1]")
@@ -128,7 +128,9 @@ public class SparkFileIndexITCase extends SparkWriteITCase {
tableRoot,
RowType.of(),
new CoreOptions(new Options()).partitionDefaultName(),
- CoreOptions.FILE_FORMAT.defaultValue().toString());
+ CoreOptions.FILE_FORMAT.defaultValue().toString(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue());
Table table = fileSystemCatalog.getTable(Identifier.create("db", "T"));
ReadBuilder readBuilder = table.newReadBuilder();
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteITCase.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteITCase.java
index 6d42f614f..0cc17639f 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteITCase.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteITCase.java
@@ -18,20 +18,28 @@
package org.apache.paimon.spark;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.io.TempDir;
+import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -41,9 +49,11 @@ public class SparkWriteITCase {
protected SparkSession spark = null;
+ protected static Path warehousePath = null;
+
@BeforeAll
public void startMetastoreAndSpark(@TempDir java.nio.file.Path tempDir) {
- Path warehousePath = new Path("file:///" + tempDir.toString());
+ warehousePath = new Path("file:///" + tempDir.toString());
spark =
SparkSession.builder()
.master("local[2]")
@@ -224,4 +234,99 @@ public class SparkWriteITCase {
rows = spark.sql("SELECT max(bucket) FROM `T$FILES`").collectAsList();
assertThat(rows.toString()).isEqualTo("[[0]]");
}
+
+ @Test
+ public void testDefaultDataFilePrefix() {
+ spark.sql("CREATE TABLE T (a INT, b INT, c STRING)");
+
+ spark.sql("INSERT INTO T VALUES (1, 1, 'aa')");
+ spark.sql("INSERT INTO T VALUES (2, 2, 'bb')");
+ spark.sql("INSERT INTO T VALUES (3, 3, 'cc')");
+
+ List<Row> data = spark.sql("SELECT * FROM T").collectAsList();
+ assertThat(data.toString()).isEqualTo("[[1,1,aa], [2,2,bb],
[3,3,cc]]");
+
+ List<Row> rows = spark.sql("select file_path from
`T$files`").collectAsList();
+ List<String> fileNames =
+ rows.stream().map(x ->
x.getString(0)).collect(Collectors.toList());
+ Assertions.assertEquals(3, fileNames.size());
+ for (String fileName : fileNames) {
+ Assertions.assertTrue(fileName.startsWith("data-"));
+ }
+ }
+
+ @Test
+ public void testDataFilePrefixForAppendOnlyTable() {
+ spark.sql("CREATE TABLE T (a INT, b INT, c STRING)");
+
+ spark.conf().set("spark.paimon.data-file.prefix", "test-");
+ spark.sql("INSERT INTO T VALUES (1, 1, 'aa')");
+ spark.sql("INSERT INTO T VALUES (2, 2, 'bb')");
+ spark.sql("INSERT INTO T VALUES (3, 3, 'cc')");
+
+ List<Row> data = spark.sql("SELECT * FROM T").collectAsList();
+ assertThat(data.toString()).isEqualTo("[[1,1,aa], [2,2,bb],
[3,3,cc]]");
+
+ List<Row> rows = spark.sql("select file_path from
`T$files`").collectAsList();
+ List<String> fileNames =
+ rows.stream().map(x ->
x.getString(0)).collect(Collectors.toList());
+ Assertions.assertEquals(3, fileNames.size());
+ for (String fileName : fileNames) {
+ Assertions.assertTrue(fileName.startsWith("test-"));
+ }
+ }
+
+ @Test
+ public void testDataFilePrefixForPKTable() {
+ spark.sql("CREATE TABLE T (a INT, b INT, c STRING)" + " TBLPROPERTIES
('primary-key'='a')");
+
+ spark.conf().set("spark.paimon.data-file.prefix", "test-");
+ spark.sql("INSERT INTO T VALUES (1, 1, 'aa')");
+ spark.sql("INSERT INTO T VALUES (2, 2, 'bb')");
+ spark.sql("INSERT INTO T VALUES (1, 3, 'cc')");
+
+ List<Row> data = spark.sql("SELECT * FROM T order by
a").collectAsList();
+ assertThat(data.toString()).isEqualTo("[[1,3,cc], [2,2,bb]]");
+
+ List<Row> rows = spark.sql("select file_path from
`T$files`").collectAsList();
+ List<String> fileNames =
+ rows.stream().map(x ->
x.getString(0)).collect(Collectors.toList());
+ Assertions.assertEquals(3, fileNames.size());
+ for (String fileName : fileNames) {
+ Assertions.assertTrue(fileName.startsWith("test-"));
+ }
+ }
+
+ @Test
+ public void testChangelogFilePrefixForPkTable() throws Exception {
+ spark.sql(
+ "CREATE TABLE T (a INT, b INT, c STRING) TBLPROPERTIES
('primary-key'='a', 'bucket' = '1', 'changelog-producer' = 'lookup')");
+
+ FileStoreTable table = getTable("T");
+ Path tabLocation = table.location();
+ FileIO fileIO = table.fileIO();
+
+ // default prefix "changelog-"
+ spark.sql("INSERT INTO T VALUES (1, 1, 'aa')");
+ FileStatus[] files1 = fileIO.listStatus(new Path(tabLocation,
"bucket-0"));
+ Assertions.assertEquals(1, dataFileCount(files1, "changelog-"));
+
+ // custom prefix "test-changelog"
+ spark.conf().set("spark.paimon.changelog-file.prefix",
"test-changelog-");
+ spark.sql("INSERT INTO T VALUES (2, 2, 'bb')");
+ FileStatus[] files2 = fileIO.listStatus(new Path(tabLocation,
"bucket-0"));
+ Assertions.assertEquals(1, dataFileCount(files2, "test-changelog-"));
+ }
+
+ protected static FileStoreTable getTable(String tableName) {
+ return FileStoreTableFactory.create(
+ LocalFileIO.create(),
+ new Path(warehousePath, String.format("db.db/%s", tableName)));
+ }
+
+ private long dataFileCount(FileStatus[] files, String filePrefix) {
+ return Arrays.stream(files)
+ .filter(f -> f.getPath().getName().startsWith(filePrefix))
+ .count();
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteWithKyroITCase.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteWithKyroITCase.java
index dd377b04c..985cd8808 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteWithKyroITCase.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkWriteWithKyroITCase.java
@@ -31,7 +31,7 @@ public class SparkWriteWithKyroITCase extends
SparkWriteITCase {
@BeforeAll
@Override
public void startMetastoreAndSpark(@TempDir java.nio.file.Path tempDir) {
- Path warehousePath = new Path("file:" + tempDir.toString());
+ warehousePath = new Path("file:" + tempDir.toString());
spark =
SparkSession.builder()
.config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
