This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 e2127c77d9 [core] Support partition-level file formats for
catalog-managed Format Tables (#10303)
e2127c77d9 is described below
commit e2127c77d950f9eccf5a77b2d9f82a2683dab7d7
Author: tonymtu <[email protected]>
AuthorDate: Thu Oct 8 16:20:12 2026 +0800
[core] Support partition-level file formats for catalog-managed Format
Tables (#10303)
---
docs/docs/concepts/rest/rest-api.md | 13 +
docs/docs/spark/format-table.md | 42 ++
docs/static/rest-catalog-open-api.yaml | 23 +-
.../rest/requests/CreatePartitionsRequest.java | 7 +-
.../format/CatalogFormatTablePartitionManager.java | 1 +
.../table/format/CatalogSplitEnumerator.java | 36 +-
.../table/format/FormatBatchWriteBuilder.java | 3 +-
.../paimon/table/format/FormatDataSplit.java | 20 +-
.../paimon/table/format/FormatReadBuilder.java | 13 +-
.../paimon/table/format/FormatTableCommit.java | 92 ++-
.../table/format/FormatTablePartitionManager.java | 5 +
.../table/format/FormatTablePartitionOptions.java | 61 ++
.../format/FormatTablePartitionStatsCollector.java | 82 ++-
.../paimon/table/format/SplitEnumerator.java | 26 +-
.../apache/paimon/rest/MockRESTCatalogTest.java | 6 +-
.../paimon/rest/RESTCatalogPartitionSupport.java | 49 ++
.../org/apache/paimon/rest/RESTCatalogServer.java | 4 +
.../table/FormatTableCommitCompatibilityTest.java | 121 ++++
.../paimon/table/format/FormatDataSplitTest.java | 22 +-
.../paimon/table/format/FormatReadBuilderTest.java | 3 +-
.../format/FormatTableCommitStatisticsTest.java | 8 +-
.../paimon/table/format/FormatTableCommitTest.java | 32 +-
.../format/FormatTablePartitionFormatTest.java | 616 +++++++++++++++++++++
.../paimon/table/format/FormatTableScanTest.java | 1 +
.../resources/compatibility/format-data-split-v3 | Bin 0 -> 576 bytes
paimon-flink/paimon-flink-common/pom.xml | 16 +
.../paimon/flink/source/FormatTableITCase.java | 85 +++
.../format/mosaic/FormatTableMosaicReadTest.java | 31 ++
...PaimonAnalyzeFormatTablePartitionsCommand.scala | 12 +-
.../format/FormatTablePartitionRepairTest.java | 39 +-
.../paimon/spark/table/PaimonFormatTableTest.scala | 54 ++
31 files changed, 1427 insertions(+), 96 deletions(-)
diff --git a/docs/docs/concepts/rest/rest-api.md
b/docs/docs/concepts/rest/rest-api.md
index 18e54faea4..b3279b2652 100644
--- a/docs/docs/concepts/rest/rest-api.md
+++ b/docs/docs/concepts/rest/rest-api.md
@@ -74,6 +74,19 @@ dropped, no data is deleted, and the request needs
`replaceStatistics=true` with
invalid, so a server that does not implement this rejects the request rather
than storing it. A
server also rejects additive statistics for a partition that already has a
custom location.
+Format Table partitions can also override the table format with `file.format`,
for example
+`"partitionOptions": [{"file.format": "orc"}]`. An explicit format updates an
existing partition
+even with `ignoreIfExists=true`, preserving other options. Omitting it
preserves the stored
+override; a partition without an override inherits the table format,
defaulting to `parquet`.
+An empty or unsupported format is rejected. The server and all readers and
writers must support
+partition formats before mixed-format partitions are used.
+
+An additive statistics request that explicitly supplies a different format
from the existing
+partition's effective format is rejected. Replacement statistics
(`replaceStatistics=true`) can
+update the format and statistics together. Each request batch is atomic for
catalog metadata;
+multiple batches and filesystem writes are not one transaction. See the
+[Format Table write and recovery
boundaries](../../spark/format-table.md#partition-file-formats).
+
## Semantic views
See [Semantic Views](semantic-views.md) for experimental definition
management, complete-document
diff --git a/docs/docs/spark/format-table.md b/docs/docs/spark/format-table.md
index 8422d703f4..70e3ce6aa9 100644
--- a/docs/docs/spark/format-table.md
+++ b/docs/docs/spark/format-table.md
@@ -112,5 +112,47 @@ change the table's managed/external ownership.
## Writes and Maintenance
+### Partition File Formats
+
+Catalog-managed partitions can override the table's `file.format` through the
partition `options`
+map in the Catalog API or REST `partitionOptions` request. For example, an ORC
partition of a
+Parquet table can be registered with `{"file.format": "orc"}`. Catalog
implementations supporting
+this feature must also apply an explicit format update to an existing
partition, including when
+`ignoreIfExists=true`, while preserving its other options. Omitting the format
leaves it unchanged.
+
+The effective format is the partition's explicit value, then the table's
value, then the default
+`parquet`. An absent partition option inherits the table format; an empty or
invalid value is an
+error. Supported formats match Format Table's table-level formats: ORC,
Parquet, CSV, TEXT, JSON
+and MOSAIC, with the same schema and dependency requirements. Other format
settings, such as CSV
+delimiters, come from the table. All files within one partition must use its
effective format.
+
+The shared Paimon reader uses this metadata in both Spark and Flink. A
partition can specify both
+`path` and `file.format`. Directory-discovered partitions and Spark's `engine`
implementation do
+not use these overrides. Updating an option describes the existing files; it
does not convert them.
+
+Writers continue to use the table's write format. Appending to a partition
registered with a
+different format fails before files are published. Overwrite replaces the
targeted partitions
+using the write format and always reports that format with the replacement
statistics. This stores
+an explicit `file.format` even when the partition previously inherited the
table format. Appends
+leave the format option unchanged, and TRUNCATE preserves the existing format.
+Changing only the table's default does not convert existing data: register the
actual format of
+partitions that would otherwise inherit the new default before changing it.
All readers and writers
+of a mixed table, and its catalog provider, must support partition file
formats.
+
+Overwrites of the same partition must be externally serialized with other
writes. Changes to a
+partition's format metadata must also be serialized with both appends and
overwrites: the append
+check does not prevent the format from changing before files are published.
Queries running during
+an overwrite have no snapshot isolation and may observe files and metadata
from different stages
+of the overwrite.
+
+If new files are published but the catalog update fails, an overwrite can
leave the previous
+format and statistics describing the replacement files. The new files are
preserved; the old
+files are not restored. With other writes stopped, verify that the replacement
is complete and
+uses one format, then repair the partition's `file.format` and replace its
statistics through the
+Catalog API. An incomplete replacement needs a fresh overwrite from the source
data.
+
+`ANALYZE TABLE` chooses the statistics reader per partition. Formats without
footer row counts
+leave the row count unknown. The existing restriction on analyzing custom
locations still applies.
+
See [SQL Writes](./sql-write#insert-overwrite) for overwrite behavior and
[TRUNCATE TABLE](./sql-write#truncate-table) for how data, registrations, and
statistics are updated.
diff --git a/docs/static/rest-catalog-open-api.yaml
b/docs/static/rest-catalog-open-api.yaml
index 02ee6df595..a52be685d0 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/docs/static/rest-catalog-open-api.yaml
@@ -986,7 +986,8 @@ paths:
summary: Create partitions
description: >
Creates partitions with optional position-aligned options. The server
validates and stores
- the partitions, options, and statistics atomically.
+ the partitions, options, and statistics atomically within each request
batch. Multiple
+ batches and filesystem writes are not one transaction.
operationId: createPartitions
parameters:
- name: prefix
@@ -2776,7 +2777,25 @@ components:
description: Whether partitionStatistics replace the stored values
rather than add to them; required whenever partitionStatistics is present, and
absent otherwise. Replacing overwrites recordCount, fileSizeInBytes, fileCount
and lastFileCreationTime; adding sums the three counts and keeps the later
lastFileCreationTime, since two timestamps do not add. A field reported as
unknown leaves the stored one alone either way, and totalBuckets is never
combined. A client that reports o [...]
type: [ boolean, "null" ]
partitionOptions:
- description: Optional options aligned with partitionSpecs by
position. The list must have the same length as partitionSpecs; use an empty
object when a partition has no options. A custom location is stored under the
path key. Omitting path keeps the stored location; a path outside the table
location registers a custom location for a new partition and must equal the
stored path for an existing one; the partition's own default directory under
the table location returns the partit [...]
+ description: >-
+ Optional options aligned with partitionSpecs by position. The list
must have the same
+ length as partitionSpecs; use an empty object when a partition has
no options.
+ A custom location is stored under the path key. Omitting path
keeps the stored location;
+ a path outside the table location registers a custom location for
a new partition and
+ must equal the stored path for an existing one; the partition's
own default directory
+ under the table location returns the partition there, dropping the
stored path without
+ deleting data. Returning a partition to its default directory
requires
+ replaceStatistics=true and a partitionStatistics entry with the
same spec, applied in
+ the same request. Any other path under the table location is
invalid, so a provider
+ that does not implement this rejects the request instead of
storing it.
+ For Format Tables, file.format overrides the table format (default
parquet). Supported
+ values are orc, parquet, csv, text, json and mosaic,
case-insensitive; empty or unsupported
+ values are rejected. An explicit format updates existing
partitions even with
+ ignoreIfExists=true, preserving other options; omitting it
preserves the stored override.
+ An additive statistics request with an explicit format differing
from the existing
+ partition's effective format is rejected. Replacement statistics
allow the format and
+ statistics to change together, atomically within the request
batch. Mixed-format
+ partitions require compatible servers, readers and writers.
type: [ array, "null" ]
items:
type: object
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
index e5260a8dcb..e9aa6eb84f 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
@@ -42,7 +42,12 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
* PartitionStatistics#spec()} rather than by position, so they may cover only
some of them. Both
* statistics fields are absent unless the client reports. Partition options
align with {@code
* partitionSpecs} by position; naming the partition's own default directory
in {@code path} asks
- * the catalog to put it back there, which needs replacement statistics for
that partition.
+ * the catalog to put it back there, which needs replacement statistics for
that partition. An
+ * explicit {@code file.format} sets the partition's format without changing
other options; omitting
+ * it preserves the stored override. Catalogs supporting partition formats
apply explicit format
+ * updates even when {@code ignoreIfExists=true}, together with the statistics
in the same batch.
+ * Additive statistics must describe files in the partition's effective
format, whereas a
+ * replacement may report a new format.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
public class CreatePartitionsRequest implements RESTRequest {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
index 72e23a200c..39f061512a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
@@ -218,6 +218,7 @@ class CatalogFormatTablePartitionManager implements
FormatTablePartitionManager
for (int i = 0; i < partitionOptions.size(); i++) {
Map<String, String> options = partitionOptions.get(i);
checkArgument(options != null, "Partition options must not contain
null maps.");
+ FormatTablePartitionOptions.fileFormatOverride(options);
checkArgument(
options.entrySet().stream()
.noneMatch(entry -> entry.getKey() == null ||
entry.getValue() == null),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
index a65d224de3..61392cc570 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
@@ -43,6 +43,7 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -89,7 +90,8 @@ final class CatalogSplitEnumerator extends SplitEnumerator {
List<Split> enumeratePartitions(@Nullable PartitionPredicate
partitionFilter)
throws IOException {
CatalogPartitionListing listing =
findCatalogPartitions(partitionFilter);
- return enumeratePartitions(filterPartitions(listing.partitionPaths,
partitionFilter));
+ return enumeratePartitions(
+ filterPartitions(listing.partitionPaths, partitionFilter),
listing.formats);
}
@Override
@@ -101,11 +103,13 @@ final class CatalogSplitEnumerator extends
SplitEnumerator {
List<Pair<LinkedHashMap<String, String>, Path>> selected =
filterPartitions(listing.partitionPaths, partitionFilter);
List<PartitionEntry> entries = toPartitionEntries(listing.partitions,
partitionFilter);
- return new ScanPlan(enumeratePartitions(selected), rowCount(entries));
+ return new ScanPlan(enumeratePartitions(selected, listing.formats),
rowCount(entries));
}
private List<Split> enumeratePartitions(
- List<Pair<LinkedHashMap<String, String>, Path>> partitions) throws
IOException {
+ List<Pair<LinkedHashMap<String, String>, Path>> partitions,
+ Map<Map<String, String>, String> formats)
+ throws IOException {
List<Split> splits = new ArrayList<>();
if (partitions.isEmpty()) {
return splits;
@@ -140,7 +144,8 @@ final class CatalogSplitEnumerator extends SplitEnumerator {
fileIOResolver.fileIO(useCatalogContextFileIO),
pair.getValue(),
partitionRow,
- useCatalogContextFileIO);
+ useCatalogContextFileIO,
+ formats.get(pair.getKey()));
} catch (FileNotFoundException e) {
warnMissingPartition(pair.getKey(), pair.getValue());
return Collections.emptyList();
@@ -179,7 +184,23 @@ final class CatalogSplitEnumerator extends SplitEnumerator
{
}
List<Pair<LinkedHashMap<String, String>, Path>> partitionPaths =
toSpecsAndPaths(partitions,
coreOptions.formatTablePartitionOnlyValueInPath());
- return new CatalogPartitionListing(partitions, partitionPaths);
+ Map<Map<String, String>, String> formats = new HashMap<>();
+ for (Partition partition : partitions) {
+ Map<String, String> spec =
+ normalizeSpec(
+ partition.spec(),
coreOptions.formatTablePartitionOnlyValueInPath());
+ String format =
+ FormatTablePartitionOptions.fileFormat(table.options(),
partition.options());
+ String previous = formats.putIfAbsent(spec, format);
+ if (previous != null && !previous.equals(format)) {
+ throw new IllegalStateException(
+ "Catalog returned conflicting file formats for
partition "
+ + spec
+ + " of Format Table "
+ + table.fullName());
+ }
+ }
+ return new CatalogPartitionListing(partitions, partitionPaths,
formats);
}
@Override
@@ -197,12 +218,15 @@ final class CatalogSplitEnumerator extends
SplitEnumerator {
private final List<Partition> partitions;
private final List<Pair<LinkedHashMap<String, String>, Path>>
partitionPaths;
+ private final Map<Map<String, String>, String> formats;
private CatalogPartitionListing(
List<Partition> partitions,
- List<Pair<LinkedHashMap<String, String>, Path>>
partitionPaths) {
+ List<Pair<LinkedHashMap<String, String>, Path>> partitionPaths,
+ Map<Map<String, String>, String> formats) {
this.partitions = partitions;
this.partitionPaths = partitionPaths;
+ this.formats = formats;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index f9d08a5c1b..a57bf1fb97 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -100,7 +100,8 @@ public class FormatBatchWriteBuilder implements
BatchWriteBuilder {
table.partitionManager(),
options.dynamicPartitionOverwrite(),
cleanupThreadNum,
- publishThreadNum);
+ publishThreadNum,
+ FormatTablePartitionOptions.fileFormat(table.options(), null));
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
index 614780e425..eebfe60be6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
@@ -40,16 +40,21 @@ public class FormatDataSplit implements Split {
private final List<FileMeta> files;
@Nullable private final BinaryRow partition;
private final boolean useCatalogContextFileIO;
+ @Nullable private final String fileFormat;
public FormatDataSplit(List<FileMeta> files, @Nullable BinaryRow
partition) {
- this(files, partition, false);
+ this(files, partition, false, null);
}
public FormatDataSplit(
- List<FileMeta> files, @Nullable BinaryRow partition, boolean
useCatalogContextFileIO) {
+ List<FileMeta> files,
+ @Nullable BinaryRow partition,
+ boolean useCatalogContextFileIO,
+ @Nullable String fileFormat) {
this.files = files;
this.partition = partition;
this.useCatalogContextFileIO = useCatalogContextFileIO;
+ this.fileFormat = fileFormat;
}
public List<FileMeta> files() {
@@ -69,6 +74,14 @@ public class FormatDataSplit implements Split {
return useCatalogContextFileIO;
}
+ /**
+ * The planned format, or null for splits serialized before partition
formats were supported.
+ */
+ @Nullable
+ public String fileFormat() {
+ return fileFormat;
+ }
+
/** Total bytes to read for this split, i.e. the sum of {@link
FileMeta#readSize()}. */
public long totalSize() {
return files.stream().mapToLong(FileMeta::readSize).sum();
@@ -99,13 +112,14 @@ public class FormatDataSplit implements Split {
}
FormatDataSplit that = (FormatDataSplit) o;
return useCatalogContextFileIO == that.useCatalogContextFileIO
+ && Objects.equals(fileFormat, that.fileFormat)
&& Objects.equals(files, that.files)
&& Objects.equals(partition, that.partition);
}
@Override
public int hashCode() {
- return Objects.hash(files, partition, useCatalogContextFileIO);
+ return Objects.hash(files, partition, useCatalogContextFileIO,
fileFormat);
}
/**
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 1bdc1dbb6b..9e7cb1d349 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -217,9 +217,18 @@ public class FormatReadBuilder implements ReadBuilder {
PredicateBuilder.splitAnd(filter), new
HashSet<>(table.partitionKeys()));
RowType dataRowType = getRowTypeWithoutPartition(table.rowType(),
table.partitionKeys());
RowType readRowType = getRowTypeWithoutPartition(readType,
table.partitionKeys());
+ Options fileOptions = new Options(options.toConfiguration().toMap());
+ fileOptions.set(
+ CoreOptions.FILE_FORMAT,
+ FormatTablePartitionOptions.fileFormat(
+ fileOptions.toMap(),
+ dataSplit.fileFormat() == null
+ ? null
+ : Collections.singletonMap(
+ CoreOptions.FILE_FORMAT.key(),
dataSplit.fileFormat())));
FormatReaderFactory readerFactory =
- FileFormatDiscover.of(options)
- .discover(options.formatType())
+ FileFormatDiscover.of(new CoreOptions(fileOptions))
+ .discover(fileOptions.get(CoreOptions.FILE_FORMAT))
.createReaderFactory(dataRowType, readRowType,
readFilters);
Pair<int[], RowType> partitionMapping =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index b0b0f5de45..d71ab7e258 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -102,7 +102,13 @@ public class FormatTableCommit implements BatchTableCommit
{
private final boolean dynamicPartitionOverwrite;
private final int cleanupThreadNum;
private final int publishThreadNum;
+ @Nullable private final String fileFormat;
+ /**
+ * Creates a legacy committer whose write format is unknown. Appends and
overwrites reject
+ * partitions with an explicit {@code file.format}, including values
stored by earlier
+ * overwrites. Use {@code FormatTable.newBatchWriteBuilder()} to supply
the write format.
+ */
public FormatTableCommit(
String location,
List<String> partitionKeys,
@@ -130,7 +136,8 @@ public class FormatTableCommit implements BatchTableCommit {
partitionManager,
dynamicPartitionOverwrite,
1,
- 1);
+ 1,
+ null);
}
FormatTableCommit(
@@ -147,7 +154,8 @@ public class FormatTableCommit implements BatchTableCommit {
@Nullable FormatTablePartitionManager partitionManager,
boolean dynamicPartitionOverwrite,
int cleanupThreadNum,
- int publishThreadNum) {
+ int publishThreadNum,
+ @Nullable String fileFormat) {
if (cleanupThreadNum < 1 || cleanupThreadNum > MAX_COMMIT_THREAD_NUM) {
throw new IllegalArgumentException(
String.format(
@@ -174,6 +182,12 @@ public class FormatTableCommit implements BatchTableCommit
{
this.dynamicPartitionOverwrite = dynamicPartitionOverwrite;
this.cleanupThreadNum = cleanupThreadNum;
this.publishThreadNum = publishThreadNum;
+ this.fileFormat =
+ fileFormat == null
+ ? null
+ : FormatTablePartitionOptions.fileFormatOverride(
+ Collections.singletonMap(
+ CoreOptions.FILE_FORMAT.key(),
fileFormat));
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -319,7 +333,8 @@ public class FormatTableCommit implements BatchTableCommit {
message.getCommitter().clean(this.fileIO);
}
if (reportsStatistics && overwrite) {
- reportPartitions(reportTargetSpecs, statisticsByPartition,
commitTime, overwrite);
+ reportPartitions(
+ reportTargetSpecs, statisticsByPartition, commitTime,
true, fileFormat);
} else if (partitionManager != null && !partitionSpecs.isEmpty()) {
// Register an append before reporting its additive
statistics. Registration is
// idempotent, so a failed multi-batch call can roll back
every file from this
@@ -351,7 +366,8 @@ public class FormatTableCommit implements BatchTableCommit {
markPublishedTargetsToPreserveOnAbort(messages);
if (reportsStatistics && !statisticsByPartition.isEmpty()) {
try {
- reportPartitions(partitionSpecs,
statisticsByPartition, commitTime, false);
+ reportPartitions(
+ partitionSpecs, statisticsByPartition,
commitTime, false, null);
} catch (RuntimeException statisticsFailure) {
LOG.warn(
"Committed data for format table {}, but
failed to report append "
@@ -417,11 +433,35 @@ public class FormatTableCommit implements
BatchTableCommit {
try {
List<Partition> targetPartitions =
loadCommitTargetPartitions(writtenPartitionSpecs);
- if (!overwrite) {
+ if (!overwrite || fileFormat == null) {
for (Partition partition : targetPartitions) {
- if
(FormatTablePartitionPathResolver.customLocation(partition) != null) {
+ if (!overwrite
+ &&
FormatTablePartitionPathResolver.customLocation(partition) != null) {
throw unsupportedCustomLocation("Writing", partition);
}
+ String partitionFormat =
+
FormatTablePartitionOptions.fileFormatOverride(partition.options());
+ if (partitionFormat != null && fileFormat == null) {
+ throw new UnsupportedOperationException(
+ "Cannot write to partition "
+ + partition.spec()
+ + " of Format Table "
+ + tableIdentifier
+ + " with an explicit file.format
because the write format is"
+ + " unknown. Use
FormatTable.newBatchWriteBuilder().");
+ }
+ if (partitionFormat != null &&
!partitionFormat.equals(fileFormat)) {
+ throw new UnsupportedOperationException(
+ "Cannot append files in format "
+ + fileFormat
+ + " to partition "
+ + partition.spec()
+ + " of Format Table "
+ + tableIdentifier
+ + " registered with file.format="
+ + partitionFormat
+ + ".");
+ }
}
}
return targetPartitions;
@@ -437,14 +477,21 @@ public class FormatTableCommit implements
BatchTableCommit {
Set<Map<String, String>> writtenPartitionSpecs) {
if (overwrite) {
if (staticPartitions == null || staticPartitions.isEmpty()) {
- return replacesOnlyWrittenPartitions()
- ? Collections.emptyList()
- : loadPartitionRegistry();
+ if (!replacesOnlyWrittenPartitions()) {
+ return loadPartitionRegistry();
+ }
+ } else {
+ LinkedHashMap<String, String> staticSpec =
orderedPartitionPrefix(staticPartitions);
+ if (staticSpec.size() < partitionKeys.size()) {
+ return loadPartitionsByPrefix(staticSpec);
+ }
+ if (fileFormat == null) {
+ return
loadPartitionsByNames(Collections.singleton(staticSpec));
+ }
+ }
+ if (fileFormat != null) {
+ return Collections.emptyList();
}
- LinkedHashMap<String, String> staticSpec =
orderedPartitionPrefix(staticPartitions);
- return staticSpec.size() == partitionKeys.size()
- ? Collections.emptyList()
- : loadPartitionsByPrefix(staticSpec);
}
return writtenPartitionSpecs.isEmpty()
@@ -585,7 +632,8 @@ public class FormatTableCommit implements BatchTableCommit {
Set<Map<String, String>> targetPartitionSpecs,
Map<Map<String, String>, PartitionStatistics>
statisticsByPartition,
long commitTime,
- boolean replaceStatistics) {
+ boolean replaceStatistics,
+ @Nullable String writtenFileFormat) {
// Statistics are matched by spec, not by position: the specs need
only be a superset.
Set<Map<String, String>> specs = new
LinkedHashSet<>(targetPartitionSpecs);
specs.addAll(statisticsByPartition.keySet());
@@ -608,8 +656,14 @@ public class FormatTableCommit implements BatchTableCommit
{
// require a fully qualified URI. Keep escaped partition
values intact.
partitionPath = new Path(new
File(partitionPath.toString()).toURI());
}
- partitionOptions.add(
- Collections.singletonMap(CoreOptions.PATH.key(),
partitionPath.toString()));
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put(CoreOptions.PATH.key(), partitionPath.toString());
+ if (writtenFileFormat != null) {
+ // Report the format of the replacement even when the
registry previously
+ // matched it. That earlier value may have changed since
the lookup.
+ options.put(CoreOptions.FILE_FORMAT.key(),
writtenFileFormat);
+ }
+ partitionOptions.add(options);
}
}
partitionManager.createPartitions(
@@ -1314,7 +1368,11 @@ public class FormatTableCommit implements
BatchTableCommit {
// too, so the catalog stops describing files that are gone.
try {
reportPartitions(
- emptied.keySet(), emptied, truncateTime, /*
replaceStatistics */ true);
+ emptied.keySet(),
+ emptied,
+ truncateTime,
+ /* replaceStatistics */ true,
+ null);
} catch (RuntimeException e) {
if (failure == null) {
throw e;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
index aebbcfd1fb..0053bbda57 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
@@ -85,6 +85,11 @@ public interface FormatTablePartitionManager extends
Serializable {
* and may split the request into catalog-sized batches; each batch is
atomic, the whole call is
* not.
*
+ * <p>A partition's {@code file.format} overrides the table format.
Supplying it explicitly
+ * updates that partition's format even when {@code ignoreIfExists=true};
omitting it preserves
+ * the stored override. The format and statistics are updated in the same
batch. A catalog must
+ * reject additive statistics whose supplied format differs from the
registered format.
+ *
* <p>This is the method an implementation provides, so that none can
report nothing by
* accident: a decorator that forwards only the two-argument form would
otherwise drop every
* report and leave the caller no way to notice.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionOptions.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionOptions.java
new file mode 100644
index 0000000000..da4d46b258
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionOptions.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FormatTable;
+
+import javax.annotation.Nullable;
+
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Resolves the file format stored in a Format Table partition's options. */
+public final class FormatTablePartitionOptions {
+
+ private FormatTablePartitionOptions() {}
+
+ public static String fileFormat(
+ Map<String, String> tableOptions, @Nullable Map<String, String>
partitionOptions) {
+ String override = fileFormatOverride(partitionOptions);
+ if (override != null) {
+ return override;
+ }
+ String tableFormat = fileFormatOverride(tableOptions);
+ return tableFormat == null ? CoreOptions.FILE_FORMAT.defaultValue() :
tableFormat;
+ }
+
+ /**
+ * Returns only an explicit override; defaults must be applied after
inheriting table options.
+ */
+ @Nullable
+ public static String fileFormatOverride(@Nullable Map<String, String>
options) {
+ if (options == null ||
!options.containsKey(CoreOptions.FILE_FORMAT.key())) {
+ return null;
+ }
+ String value = options.get(CoreOptions.FILE_FORMAT.key());
+ checkArgument(
+ value != null && !value.trim().isEmpty(),
+ "Format Table option '%s' must not be null or empty.",
+ CoreOptions.FILE_FORMAT.key());
+ return FormatTable.parseFormat(value).name().toLowerCase(Locale.ROOT);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java
index 09f1eceea5..3d941df50a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionStatsCollector.java
@@ -23,6 +23,8 @@ import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.SimpleStatsExtractor;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.table.FormatTable;
@@ -40,6 +42,7 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -104,23 +107,68 @@ public class FormatTablePartitionStatsCollector {
* caller can send it straight to the catalog alongside the same specs.
*/
public List<PartitionStatistics> collect(List<Map<String, String>>
partitions) {
+ List<Partition> registered =
+ partitions.isEmpty() || table.partitionManager() == null
+ ? Collections.emptyList()
+ :
table.partitionManager().listPartitionsByNames(partitions);
+ return collect(partitions, partitionFormats(registered));
+ }
+
+ /**
+ * Measures the same partition metadata that the caller planned, without
reloading the catalog.
+ */
+ public List<PartitionStatistics> collectPartitions(List<Partition>
partitions) {
+ List<Map<String, String>> specs = new ArrayList<>(partitions.size());
+ for (Partition partition : partitions) {
+ specs.add(partition.spec());
+ }
+ return collect(specs, partitionFormats(partitions));
+ }
+
+ private Map<Map<String, String>, String> partitionFormats(List<Partition>
partitions) {
+ Map<Map<String, String>, String> formats = new HashMap<>();
+ for (Partition partition : partitions) {
+ if (FormatTablePartitionPathResolver.customLocation(partition) !=
null) {
+ throw new UnsupportedOperationException(
+ "Cannot measure a Format Table partition with a custom
location: "
+ + partition.spec());
+ }
+ String format =
+ FormatTablePartitionOptions.fileFormat(table.options(),
partition.options());
+ String previous = formats.putIfAbsent(partition.spec(), format);
+ if (previous != null && !previous.equals(format)) {
+ throw new IllegalStateException(
+ "Conflicting file formats for partition " +
partition.spec());
+ }
+ }
+ return formats;
+ }
+
+ private List<PartitionStatistics> collect(
+ List<Map<String, String>> partitions, Map<Map<String, String>,
String> formats) {
if (partitions.isEmpty()) {
return Collections.emptyList();
}
- SimpleStatsExtractor rowCounter = withRecordCount ? rowCounter() :
null;
- if (withRecordCount && rowCounter == null) {
- LOG.info(
- "No row counter could be built for format {} of table {},
so the row counts of "
- + "the measured partitions stay unknown.",
- table.format(),
- table.fullName());
+ String defaultFormat =
FormatTablePartitionOptions.fileFormat(table.options(), null);
+ Map<String, Optional<SimpleStatsExtractor>> counters = new HashMap<>();
+ List<SimpleStatsExtractor> rowCounters = new
ArrayList<>(partitions.size());
+ for (Map<String, String> partition : partitions) {
+ String format = formats.getOrDefault(partition, defaultFormat);
+ rowCounters.add(
+ withRecordCount
+ ? counters.computeIfAbsent(
+ format, f ->
Optional.ofNullable(rowCounter(f)))
+ .orElse(null)
+ : null);
}
// A listing is one request per partition, a footer read one per file,
so counting rows
// leaves work to spread even when a single partition was asked for.
- int threads = rowCounter == null ? Math.min(parallelism,
partitions.size()) : parallelism;
+ int threads = withRecordCount ? parallelism : Math.min(parallelism,
partitions.size());
if (threads == 1) {
List<PartitionStatistics> statistics = new
ArrayList<>(partitions.size());
- for (Map<String, String> partition : partitions) {
+ for (int i = 0; i < partitions.size(); i++) {
+ Map<String, String> partition = partitions.get(i);
+ SimpleStatsExtractor rowCounter = rowCounters.get(i);
List<FileStatus> files = listDataFiles(partition);
List<Long> rowCounts = new ArrayList<>(files.size());
for (FileStatus file : files) {
@@ -147,7 +195,9 @@ public class FormatTablePartitionStatsCollector {
// Every file of every partition goes to the same pool, so one
partition holding many
// files is counted with all of it rather than with one thread of
it.
List<List<Future<Long>>> rowCounts = new
ArrayList<>(partitions.size());
- for (List<FileStatus> partitionFiles : files) {
+ for (int i = 0; i < files.size(); i++) {
+ List<FileStatus> partitionFiles = files.get(i);
+ SimpleStatsExtractor rowCounter = rowCounters.get(i);
List<Future<Long>> counts = new
ArrayList<>(partitionFiles.size());
for (FileStatus file : partitionFiles) {
if (rowCounter != null) {
@@ -164,7 +214,9 @@ public class FormatTablePartitionStatsCollector {
}
statistics.add(
statistics(
- partitions.get(i), files.get(i), sum(counted,
rowCounter != null)));
+ partitions.get(i),
+ files.get(i),
+ sum(counted, rowCounters.get(i) != null)));
}
return statistics;
} finally {
@@ -271,9 +323,11 @@ public class FormatTablePartitionStatsCollector {
* table's.
*/
@Nullable
- private SimpleStatsExtractor rowCounter() {
+ private SimpleStatsExtractor rowCounter(String fileFormat) {
try {
- CoreOptions options = new CoreOptions(table.options());
+ Options fileOptions = new Options(table.options());
+ fileOptions.set(CoreOptions.FILE_FORMAT, fileFormat);
+ CoreOptions options = new CoreOptions(fileOptions);
Optional<SimpleStatsExtractor> extractor =
FileFormat.fileFormat(options).createStatsExtractor(NO_COLUMNS, NO_COLLECTORS);
return extractor.orElse(null);
@@ -281,7 +335,7 @@ public class FormatTablePartitionStatsCollector {
LOG.warn(
"Failed to create a row counter for format {} of table {};
row counts stay "
+ "unknown.",
- table.format(),
+ fileFormat,
table.fullName(),
e);
return null;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
index b6578fd859..882551c833 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
@@ -63,14 +63,12 @@ abstract class SplitEnumerator {
protected final CoreOptions coreOptions;
protected final long targetSplitSize;
protected final long openFileCost;
- protected final FormatTable.Format format;
SplitEnumerator(FormatTable table, CoreOptions coreOptions) {
this.table = table;
this.coreOptions = coreOptions;
this.targetSplitSize = coreOptions.splitTargetSize();
this.openFileCost = coreOptions.splitOpenFileCost();
- this.format = table.format();
}
static SplitEnumerator create(
@@ -145,21 +143,28 @@ abstract class SplitEnumerator {
List<Split> createSplits(FileIO fileIO, Path path, @Nullable BinaryRow
partition)
throws IOException {
- return createSplits(fileIO, path, partition, false);
+ return createSplits(
+ fileIO,
+ path,
+ partition,
+ false,
+ FormatTablePartitionOptions.fileFormat(table.options(), null));
}
List<Split> createSplits(
FileIO fileIO,
Path path,
@Nullable BinaryRow partition,
- boolean useCatalogContextFileIO)
+ boolean useCatalogContextFileIO,
+ String fileFormat)
throws IOException {
+ FormatTable.Format partitionFormat =
FormatTable.parseFormat(fileFormat);
List<FormatDataSplit.FileMeta> segments = new ArrayList<>();
// The listed directory is a single partition, or the table itself
when unpartitioned.
List<FileStatus> files = FormatTableScan.listDataFiles(fileIO, path);
files.sort(Comparator.comparing(file -> file.getPath().toString()));
for (FileStatus file : files) {
- segments.addAll(toSegments(file));
+ segments.addAll(toSegments(file, partitionFormat));
}
List<Split> splits = new ArrayList<>();
@@ -168,13 +173,14 @@ abstract class SplitEnumerator {
segments,
file -> Math.max(file.readSize(), openFileCost),
targetSplitSize)) {
- splits.add(new FormatDataSplit(bin, partition,
useCatalogContextFileIO));
+ splits.add(new FormatDataSplit(bin, partition,
useCatalogContextFileIO, fileFormat));
}
return splits;
}
- private List<FormatDataSplit.FileMeta> toSegments(FileStatus file) {
- if (!preferToSplitFile(file)) {
+ private List<FormatDataSplit.FileMeta> toSegments(
+ FileStatus file, FormatTable.Format fileFormat) {
+ if (!preferToSplitFile(file, fileFormat)) {
return Collections.singletonList(
new FormatDataSplit.FileMeta(file.getPath(),
file.getLen()));
}
@@ -193,13 +199,13 @@ abstract class SplitEnumerator {
return segments;
}
- private boolean preferToSplitFile(FileStatus file) {
+ private boolean preferToSplitFile(FileStatus file, FormatTable.Format
fileFormat) {
if (file.getLen() <= targetSplitSize) {
return false;
}
Options options = coreOptions.toConfiguration();
- switch (format) {
+ switch (fileFormat) {
case CSV:
return !isCompressed(file.getPath())
&&
isDefaultDelimiter(options.get(CsvOptions.LINE_DELIMITER));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index 1b82f86cd6..6e5c96ccf1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -909,7 +909,11 @@ class MockRESTCatalogTest extends RESTCatalogTest {
assertThat(partition.recordCount()).isZero();
assertThat(partition.fileSizeInBytes()).isZero();
assertThat(partition.fileCount()).isZero();
- assertThat(partition.options()).isEqualTo(partitionOptions);
+ Map<String, String> expectedOptions = new HashMap<>(partitionOptions);
+ if (operation.equals("overwrite")) {
+ expectedOptions.put(CoreOptions.FILE_FORMAT.key(), "parquet");
+ }
+ assertThat(partition.options()).isEqualTo(expectedOptions);
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
index 28de8abd30..af898f52c6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
+++
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
@@ -28,6 +28,7 @@ import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.partition.PartitionUtils;
import org.apache.paimon.rest.requests.CreatePartitionsRequest;
import org.apache.paimon.rest.responses.ErrorResponse;
+import org.apache.paimon.table.format.FormatTablePartitionOptions;
import org.apache.paimon.table.format.FormatTablePartitionPathResolver;
import org.apache.paimon.table.format.FormatTablePartitionRegistryValidator;
import org.apache.paimon.utils.StringUtils;
@@ -94,6 +95,10 @@ final class RESTCatalogPartitionSupport {
for (int i = 0; i < options.size(); i++) {
Map<String, String> partitionOptions = options.get(i);
Map<String, String> copied = new HashMap<>(partitionOptions);
+ String format =
FormatTablePartitionOptions.fileFormatOverride(copied);
+ if (format != null) {
+ copied.put(CoreOptions.FILE_FORMAT.key(), format);
+ }
String location = copied.get(PATH.key());
if (location != null) {
// What a partition may own is judged once the location is
known not to be the
@@ -168,6 +173,50 @@ final class RESTCatalogPartitionSupport {
return copied;
}
+ static void applyRequestedFormats(
+ List<Partition> stored,
+ CreatePartitionsRequest request,
+ @Nullable List<Map<String, String>> options,
+ TableMetadata metadata) {
+ if (options == null) {
+ return;
+ }
+ Map<Map<String, String>, String> formats = new HashMap<>();
+ for (int i = 0; i < options.size(); i++) {
+ String format =
FormatTablePartitionOptions.fileFormatOverride(options.get(i));
+ if (format != null) {
+ formats.put(request.getPartitionSpecs().get(i), format);
+ }
+ }
+ Set<Map<String, String>> appended = new HashSet<>();
+ if (!Boolean.TRUE.equals(request.replaceStatistics())
+ && request.getPartitionStatistics() != null) {
+ for (PartitionStatistics statistic :
request.getPartitionStatistics()) {
+ appended.add(statistic.spec());
+ }
+ }
+ for (int i = 0; i < stored.size(); i++) {
+ Partition partition = stored.get(i);
+ String format = formats.get(partition.spec());
+ if (format == null) {
+ continue;
+ }
+ if (appended.contains(partition.spec())
+ && !format.equals(
+ FormatTablePartitionOptions.fileFormat(
+ metadata.schema().options(),
partition.options()))) {
+ throw new IllegalArgumentException(
+ "Cannot append a different file format to partition "
+ partition.spec());
+ }
+ Map<String, String> updated =
+ partition.options() == null
+ ? new HashMap<>()
+ : new HashMap<>(partition.options());
+ updated.put(CoreOptions.FILE_FORMAT.key(), format);
+ stored.set(i, copyPartition(partition, updated));
+ }
+ }
+
private static void validateNoAdditiveStatisticsForCustomPartitions(
List<Partition> stored,
@Nullable List<PartitionStatistics> statistics,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index d196e3be23..b4afe6adaf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -2270,6 +2270,10 @@ public class RESTCatalogServer {
existed.add(spec);
}
}
+ if (formatTable) {
+ RESTCatalogPartitionSupport.applyRequestedFormats(
+ storedPartitions, request, requestedOptions,
tableMetadata);
+ }
RESTCatalogPartitionSupport.settlePartitionLocations(
storedPartitions,
returningToDefault,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
index 632ea63c46..595d13ae3f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
@@ -20,13 +20,36 @@ package org.apache.paimon.table;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.RenamingTwoPhaseOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.partition.Partition;
import org.apache.paimon.table.format.FormatTableCommit;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
+import org.apache.paimon.table.format.TwoPhaseCommitMessage;
+import org.apache.paimon.table.sink.CommitMessage;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Compatibility tests for the public Format Table commit API. */
class FormatTableCommitCompatibilityTest {
@@ -53,4 +76,102 @@ class FormatTableCommitCompatibilityTest {
commit.commit(Collections.emptyList());
}
}
+
+ @ParameterizedTest
+ @CsvSource({
+ "append, false",
+ "append, true",
+ "static, false",
+ "static, true",
+ "dynamic, false",
+ "dynamic, true",
+ "prefix, false",
+ "prefix, true",
+ "table, false",
+ "table, true",
+ "empty-static, false",
+ "empty-static, true"
+ })
+ void testLegacyConstructorRequiresWriteFormatForExplicitPartitions(
+ String operation, boolean explicitFormat) throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path partitionPath = new Path(tablePath, "pt=p/sub=s");
+ fileIO.mkdirs(partitionPath);
+ Path oldFile = new Path(partitionPath, "old.parquet");
+ Path newFile = new Path(partitionPath, "new.parquet");
+ fileIO.overwriteFileUtf8(oldFile, "old");
+ List<CommitMessage> messages;
+ if (operation.equals("empty-static")) {
+ messages = Collections.emptyList();
+ } else {
+ RenamingTwoPhaseOutputStream stream =
+ new RenamingTwoPhaseOutputStream(fileIO, newFile, false);
+ stream.write("replacement".getBytes(StandardCharsets.UTF_8));
+ messages =
+ Collections.singletonList(new
TwoPhaseCommitMessage(stream.closeForCommit()));
+ }
+
+ Map<String, String> spec = new LinkedHashMap<>();
+ spec.put("pt", "p");
+ spec.put("sub", "s");
+ Partition partition =
+ new Partition(
+ spec,
+ 1,
+ 3,
+ 1,
+ 0,
+ -1,
+ false,
+ null,
+ null,
+ null,
+ null,
+ explicitFormat ?
Collections.singletonMap("file.format", "parquet") : null);
+ FormatTablePartitionManager manager =
mock(FormatTablePartitionManager.class);
+ when(manager.listPartitions(anyMap(), any()))
+ .thenReturn(Collections.singletonList(partition));
+ when(manager.listPartitionsByNames(anyList()))
+ .thenReturn(Collections.singletonList(partition));
+ Map<String, String> staticSpec = Collections.emptyMap();
+ if (operation.equals("static") || operation.equals("empty-static")) {
+ staticSpec = spec;
+ } else if (operation.equals("prefix")) {
+ staticSpec = Collections.singletonMap("pt", "p");
+ }
+
+ try (FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("pt", "sub"),
+ fileIO,
+ false,
+ "__DEFAULT_PARTITION__",
+ !operation.equals("append"),
+ Identifier.create("compatibility_db",
"compatibility_table"),
+ staticSpec,
+ null,
+ null,
+ manager,
+ operation.equals("dynamic"))) {
+ if (explicitFormat) {
+ assertThatThrownBy(() -> commit.commit(messages))
+
.hasRootCauseInstanceOf(UnsupportedOperationException.class)
+ .hasStackTraceContaining("write format is unknown")
+
.hasStackTraceContaining("FormatTable.newBatchWriteBuilder()");
+ assertThat(fileIO.readFileUtf8(oldFile)).isEqualTo("old");
+ assertThat(fileIO.exists(newFile)).isFalse();
+ verify(manager, never())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+ } else {
+ commit.commit(messages);
+
assertThat(fileIO.exists(oldFile)).isEqualTo(operation.equals("append"));
+
assertThat(fileIO.exists(newFile)).isEqualTo(!messages.isEmpty());
+ }
+ }
+ if (!messages.isEmpty()) {
+ assertThat(fileIO.listStatus(new Path(partitionPath,
"_temporary"))).isEmpty();
+ }
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
index 45211c29a1..c2330df900 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
@@ -25,6 +25,7 @@ import org.apache.paimon.utils.InstantiationUtil;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.io.InputStream;
import java.io.ObjectStreamClass;
import java.util.Arrays;
@@ -73,7 +74,8 @@ public class FormatDataSplitTest {
new FormatDataSplit(
Arrays.asList(new FileMeta(new
Path("oss://archive/data.csv"), 10L)),
null,
- true);
+ true,
+ null);
FormatDataSplit deserialized =
InstantiationUtil.deserializeObject(
@@ -83,6 +85,24 @@ public class FormatDataSplitTest {
assertThat(deserialized.useCatalogContextFileIO()).isTrue();
}
+ @Test
+ public void testDeserializeSplitWithoutFileFormat() throws Exception {
+ // Serialized by FormatDataSplit at ce0633cb09, before the fileFormat
field was added.
+ try (InputStream input =
+ getClass()
+ .getClassLoader()
+
.getResourceAsStream("compatibility/format-data-split-v3")) {
+ assertThat(input).isNotNull();
+ FormatDataSplit split =
+ InstantiationUtil.deserializeObject(input,
getClass().getClassLoader());
+ assertThat(split.fileFormat()).isNull();
+ assertThat(split.useCatalogContextFileIO()).isTrue();
+ assertThat(split.partition()).isNull();
+ assertThat(split.files())
+ .containsExactly(new FileMeta(new
Path("/legacy/file.orc"), 128L));
+ }
+ }
+
@Test
public void testSerialVersionUIDsStayCompatible() {
assertThat(ObjectStreamClass.lookup(FormatDataSplit.class).getSerialVersionUID())
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
index 3e48b5e9b2..e1b08424d2 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
@@ -310,7 +310,8 @@ public class FormatReadBuilderTest {
csvFile,
clientFileIO.getFileSize(csvFile))),
null,
- true)),
+ true,
+ null)),
getClass().getClassLoader());
List<InternalRow> rows = readAllRows(readBuilder.createReader(split),
rowType);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
index ef891793e1..b0f996191f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
@@ -188,7 +188,8 @@ class FormatTableCommitStatisticsTest {
.containsExactly(returnsToDefault(tablePath, spec("2025",
"10")));
assertThat(fileIO.exists(oldWrittenPartitionData)).isFalse();
verify(partitionManager, never()).listPartitions(any(), isNull());
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager)
+ .listPartitionsByNames(Collections.singletonList(spec("2025",
"10")));
assertThat(reported.statistics).hasSize(1);
PartitionStatistics statistics = reported.statistics.get(0);
assertThat(statistics.recordCount()).isEqualTo(3);
@@ -197,8 +198,7 @@ class FormatTableCommitStatisticsTest {
}
@Test
- void
testEmptyStaticOverwriteReportsZeroAndResetsLocationWithoutRegistryRead()
- throws Exception {
+ void testEmptyStaticOverwriteReportsZeroAndResetsLocation() throws
Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager =
mock(FormatTablePartitionManager.class);
@@ -217,7 +217,7 @@ class FormatTableCommitStatisticsTest {
assertThat(reported.partitionOptions)
.containsExactly(returnsToDefault(tablePath, spec("2025",
"10")));
verify(partitionManager, never()).listPartitions(any(), isNull());
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(target));
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
index 24d957f97a..590d0398f5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -234,7 +234,7 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(customData)).isTrue();
assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(spec));
verify(partitionManager,
never()).listPartitions(Collections.emptyMap(), null);
assertReplacementReport(partitionManager, tablePath,
Collections.singletonList(spec));
}
@@ -280,7 +280,7 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(customData)).isTrue();
assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
verify(partitionManager,
never()).listPartitions(Collections.emptyMap(), null);
assertReplacementReport(partitionManager, tablePath,
Collections.singletonList(targetSpec));
}
@@ -420,7 +420,7 @@ class FormatTableCommitTest {
assertThat(fileIO.mkdirsCalls()).isZero();
verify(committer).commit(fileIO);
verify(committer).clean(fileIO);
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(spec));
verify(partitionManager,
never()).listPartitions(Collections.emptyMap(), null);
assertReplacementReport(partitionManager, tablePath,
Collections.singletonList(spec));
}
@@ -473,7 +473,7 @@ class FormatTableCommitTest {
assertThat(fileIO.mkdirsCalls()).isZero();
verify(committer).commit(fileIO);
verify(committer).clean(fileIO);
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
verify(partitionManager,
never()).listPartitions(Collections.emptyMap(), null);
assertReplacementReport(partitionManager, tablePath,
Collections.singletonList(targetSpec));
}
@@ -615,7 +615,7 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(oldData)).isFalse();
assertThat(fileIO.exists(new Path(tablePath,
"part=default"))).isTrue();
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+
verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
verify(partitionManager,
never()).listPartitions(Collections.emptyMap(), null);
assertReplacementReport(partitionManager, tablePath,
Collections.singletonList(targetSpec));
}
@@ -2185,7 +2185,7 @@ class FormatTableCommitTest {
assertThat(thirdFinished.await(3, TimeUnit.SECONDS)).isTrue();
assertThat(activePublishes).hasValue(1);
verify(partitionManager, never())
- .createPartitions(anyList(), eq(true), any(),
anyBoolean(), isNull());
+ .createPartitions(anyList(), eq(true), any(),
anyBoolean(), any());
for (TwoPhaseOutputStream.Committer committer : committers) {
verify(committer, never()).clean(fileIO);
}
@@ -2770,7 +2770,8 @@ class FormatTableCommitTest {
partitionManager,
/* dynamicPartitionOverwrite */ true,
/* cleanupThreadNum */ 2,
- /* publishThreadNum */ 1);
+ /* publishThreadNum */ 1,
+ null);
commit.commit(Collections.emptyList());
@@ -2853,7 +2854,8 @@ class FormatTableCommitTest {
partitionManager,
/* dynamicPartitionOverwrite */ true,
/* cleanupThreadNum */ 4,
- /* publishThreadNum */ 1);
+ /* publishThreadNum */ 1,
+ null);
commit.commit(Collections.emptyList());
@@ -3069,7 +3071,8 @@ class FormatTableCommitTest {
null,
/* dynamicPartitionOverwrite */ true,
/* cleanupThreadNum */ 1,
- publishThreadNum);
+ publishThreadNum,
+ null);
}
private void assertOverwriteCleanupSpansPartitions(boolean
dynamicPartitionOverwrite)
@@ -3144,6 +3147,8 @@ class FormatTableCommitTest {
.field("part", DataTypes.STRING())
.field("id", DataTypes.INT())
.build();
+ Map<String, String> tableOptions = new LinkedHashMap<>(options);
+ tableOptions.put(CoreOptions.FILE_FORMAT.key(), "csv");
FormatTable table =
FormatTable.builder()
.fileIO(fileIO)
@@ -3152,7 +3157,7 @@ class FormatTableCommitTest {
.partitionKeys(Collections.singletonList("part"))
.location(tablePath.toString())
.format(FormatTable.Format.CSV)
- .options(options)
+ .options(tableOptions)
.partitionManager(partitionManager)
.build();
return table;
@@ -3163,6 +3168,8 @@ class FormatTableCommitTest {
FileIO fileIO,
FormatTablePartitionManager partitionManager,
Map<String, String> options) {
+ Map<String, String> tableOptions = new LinkedHashMap<>(options);
+ tableOptions.put(CoreOptions.FILE_FORMAT.key(), "csv");
FormatTable table =
FormatTable.builder()
.fileIO(fileIO)
@@ -3171,7 +3178,7 @@ class FormatTableCommitTest {
.partitionKeys(Collections.emptyList())
.location(tablePath.toString())
.format(FormatTable.Format.CSV)
- .options(options)
+ .options(tableOptions)
.partitionManager(partitionManager)
.build();
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
@@ -3199,7 +3206,8 @@ class FormatTableCommitTest {
partitionManager,
/* dynamicPartitionOverwrite */ true,
cleanupThreadNum,
- /* publishThreadNum */ 1);
+ /* publishThreadNum */ 1,
+ null);
}
private static void writeOldFiles(LocalFileIO fileIO, Path partitionPath,
int count)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionFormatTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionFormatTest.java
new file mode 100644
index 0000000000..6f49567bb5
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionFormatTest.java
@@ -0,0 +1,616 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests partition format inheritance, file planning and reads through the
shared core path. */
+class FormatTablePartitionFormatTest {
+
+ private static final RowType DATA_TYPE =
+ RowType.builder().field("value", DataTypes.STRING()).build();
+ private static final RowType TABLE_TYPE =
+ RowType.builder()
+ .field("value", DataTypes.STRING())
+ .field("pt", DataTypes.STRING())
+ .build();
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testDefaultsAndExplicitInvalidValues() {
+
assertThat(FormatTablePartitionOptions.fileFormat(Collections.emptyMap(), null))
+ .isEqualTo("parquet");
+ assertThat(FormatTablePartitionOptions.fileFormat(options("orc"),
Collections.emptyMap()))
+ .isEqualTo("orc");
+ assertThat(FormatTablePartitionOptions.fileFormat(options("orc"),
options("parquet")))
+ .isEqualTo("parquet");
+
assertThat(FormatTablePartitionOptions.fileFormat(Collections.emptyMap(),
options("ORC")))
+ .isEqualTo("orc");
+ for (String invalid : Arrays.asList(null, "", " ", "avro", "unknown"))
{
+ assertThatThrownBy(
+ () ->
+ FormatTablePartitionOptions.fileFormat(
+ options("orc"), options(invalid)))
+ .isInstanceOf(RuntimeException.class);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(FormatTable.Format.class)
+ void testEveryTableFormatCanBeInheritedOrOverridden(FormatTable.Format
format) {
+ String expected = format.name().toLowerCase(java.util.Locale.ROOT);
+
assertThat(FormatTablePartitionOptions.fileFormat(options(format.name()), null))
+ .isEqualTo(expected);
+ assertThat(
+ FormatTablePartitionOptions.fileFormat(
+ options("parquet"), options(format.name())))
+ .isEqualTo(expected);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet", "csv", "json", "text"})
+ void testMixedPartitionReadsAndStatistics(String override) throws
Exception {
+ String inherited = override.equals("orc") ? "parquet" : "orc";
+ Path tablePath = new Path(tempDir.toUri());
+ writeFile(tablePath, "inherited", inherited, "base");
+ writeFile(tablePath, "overridden", override, "first", "second");
+ List<Partition> partitions =
+ Arrays.asList(
+ partition("inherited", null), partition("overridden",
options(override)));
+ FormatTable table = table(tablePath, inherited, partitions);
+
+ ReadBuilder read = table.newReadBuilder();
+ List<Split> splits = read.newScan().plan().splits();
+ assertThat(splits).hasSize(2);
+ for (Split split : splits) {
+ FormatDataSplit copy = InstantiationUtil.clone((FormatDataSplit)
split);
+ assertThat(copy).isEqualTo(split);
+ assertThat(copy.fileFormat()).isIn(inherited, override);
+ }
+ assertThat(readRows(read, splits)).containsExactlyInAnyOrder("base",
"first", "second");
+
+ ReadBuilder selected =
+ table.newReadBuilder()
+ .withFilter(
+ new PredicateBuilder(TABLE_TYPE)
+ .equal(1,
BinaryString.fromString("overridden")))
+ .withReadType(DATA_TYPE);
+ assertThat(readRows(selected, selected.newScan().plan().splits()))
+ .containsExactly("first", "second");
+
+ List<PartitionStatistics> statistics =
+ new FormatTablePartitionStatsCollector(table, true, 2)
+ .collectPartitions(partitions);
+ assertThat(statistics.get(0).recordCount()).isEqualTo(1);
+ assertThat(statistics.get(1).recordCount())
+ .isEqualTo(
+ override.equals("orc") || override.equals("parquet")
+ ? 2
+ : PartitionStatistics.UNKNOWN);
+ assertThat(statistics).allSatisfy(stat ->
assertThat(stat.fileCount()).isEqualTo(1));
+
+ // A legacy split must inherit ORC, rather than accidentally taking
the global Parquet
+ // default.
+ if (inherited.equals("orc")) {
+ FormatDataSplit base =
+ (FormatDataSplit)
+ splits.stream()
+ .filter(
+ split ->
+ ((FormatDataSplit) split)
+ .fileFormat()
+ .equals("orc"))
+ .findFirst()
+ .get();
+ FormatDataSplit legacy = new FormatDataSplit(base.files(),
base.partition());
+ assertThat(readRows(read,
Collections.singletonList(legacy))).containsExactly("base");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"00000_0", "part-x.parquet.gz"})
+ void testPartitionFormatTakesPrecedenceOverFileName(String fileName)
throws Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "orc", "first", "second");
+ FormatTable table =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(partition("p",
options("orc"))));
+ assertThat(
+ table.fileIO()
+ .rename(
+ new Path(location, "pt=p/data.orc"),
+ new Path(location, "pt=p/" +
fileName)))
+ .isTrue();
+ ReadBuilder read = table.newReadBuilder();
+ assertThat(readRows(read, read.newScan().plan().splits()))
+ .containsExactly("first", "second");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"csv", "json"})
+ void testSplittingUsesPartitionFormat(String format) throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ writeFile(tablePath, "override", format, "first", "second", "third",
"fourth");
+ FormatTable table =
+ table(
+ tablePath,
+ "parquet",
+
Collections.singletonList(partition("override", options(format))))
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "8 b"));
+ ReadBuilder read = table.newReadBuilder();
+ List<Split> splits = read.newScan().plan().splits();
+ assertThat(splits.size()).isGreaterThan(1);
+ assertThat(readRows(read, splits)).containsExactly("first", "second",
"third", "fourth");
+ }
+
+ @Test
+ void testCustomPathAndFormatAreIndependent() throws Exception {
+ Path tablePath = new Path(tempDir.resolve("table").toUri());
+ Path external = new Path(tempDir.resolve("external").toUri());
+ writeFile(external, "archive", "orc", "external");
+ Map<String, String> partitionOptions = options("orc");
+ partitionOptions.put(CoreOptions.PATH.key(), new Path(external,
"pt=archive").toString());
+ FormatTable table =
+ table(
+ tablePath,
+ "parquet",
+ Collections.singletonList(partition("archive",
partitionOptions)));
+ ReadBuilder read = table.newReadBuilder();
+ assertThat(readRows(read,
read.newScan().plan().splits())).containsExactly("external");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"static", "dynamic", "table"})
+ void testAppendRejectsDifferentFormatAndOverwriteUpdatesFormat(String
overwriteMode)
+ throws Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "orc", "old");
+ FormatTable table =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(partition("p",
options("orc"))))
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(),
+
Boolean.toString(overwriteMode.equals("dynamic"))));
+ BatchWriteBuilder append = table.newBatchWriteBuilder();
+ List<CommitMessage> appendMessages;
+ try (BatchTableWrite write = append.newWrite()) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("rejected"),
BinaryString.fromString("p")));
+ appendMessages = write.prepareCommit();
+ }
+ assertThatThrownBy(() -> append.newCommit().commit(appendMessages))
+ .hasStackTraceContaining("registered with file.format=orc");
+ assertThat(FormatTableScan.listDataFiles(table.fileIO(), new
Path(location, "pt=p")))
+ .hasSize(1);
+ assertThat(
+ readRows(
+ table.newReadBuilder(),
+
table.newReadBuilder().newScan().plan().splits()))
+ .containsExactly("old");
+
+ AtomicReference<Map<String, String>> reported = new
AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ List<Map<String, String>> options =
invocation.getArgument(4);
+ if (options != null) {
+ reported.set(options.get(0));
+ }
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+ BatchWriteBuilder overwrite =
+ table.newBatchWriteBuilder()
+ .withOverwrite(
+ overwriteMode.equals("static")
+ ? Collections.singletonMap("pt", "p")
+ : Collections.emptyMap());
+ try (BatchTableWrite write = overwrite.newWrite()) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("replacement"),
BinaryString.fromString("p")));
+ overwrite.newCommit().commit(write.prepareCommit());
+ }
+ assertThat(reported.get()).containsEntry("file.format", "parquet");
+ FormatTable reloaded =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(
+ partition("p",
options(reported.get().get("file.format")))));
+ assertThat(
+ readRows(
+ reloaded.newReadBuilder(),
+
reloaded.newReadBuilder().newScan().plan().splits()))
+ .containsExactly("replacement");
+ }
+
+ @Test
+ void testCrossFormatOverwriteCatalogFailureAndMetadataRepair() throws
Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "orc", "old");
+ List<Partition> registered =
+ new ArrayList<>(Collections.singletonList(partition("p",
options("orc"))));
+ FormatTable table = table(location, "parquet", registered);
+ doThrow(new RuntimeException("catalog update failed"))
+ .doAnswer(
+ invocation -> {
+ List<PartitionStatistics> statistics =
invocation.getArgument(2);
+ List<Map<String, String>> options =
invocation.getArgument(4);
+ PartitionStatistics stats = statistics.get(0);
+ registered.set(
+ 0,
+ new Partition(
+ stats.spec(),
+ stats.recordCount(),
+ stats.fileSizeInBytes(),
+ stats.fileCount(),
+ stats.lastFileCreationTime(),
+ stats.totalBuckets(),
+ false,
+ null,
+ null,
+ null,
+ null,
+ options.get(0)));
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+
+ BatchWriteBuilder overwrite =
+
table.newBatchWriteBuilder().withOverwrite(Collections.singletonMap("pt", "p"));
+ List<CommitMessage> messages;
+ try (BatchTableWrite write = overwrite.newWrite()) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("replacement"),
BinaryString.fromString("p")));
+ messages = write.prepareCommit();
+ }
+ assertThatThrownBy(() -> overwrite.newCommit().commit(messages))
+ .hasRootCauseMessage("catalog update failed");
+ Path replacement = ((TwoPhaseCommitMessage)
messages.get(0)).getCommitter().targetPath();
+ assertThat(table.fileIO().exists(replacement)).isTrue();
+ assertThat(table.fileIO().exists(new Path(location,
"pt=p/data.orc"))).isFalse();
+ assertThat(registered.get(0).options()).containsEntry("file.format",
"orc");
+ ReadBuilder stale = table.newReadBuilder();
+ assertThatThrownBy(() -> readRows(stale,
stale.newScan().plan().splits()))
+ .isInstanceOf(IOException.class);
+
+ // With the complete replacement verified and other writes stopped,
repair only metadata.
+ Partition repair = partition("p", options("parquet"));
+ List<PartitionStatistics> statistics =
+ new FormatTablePartitionStatsCollector(table, true, 1)
+ .collectPartitions(Collections.singletonList(repair));
+ table.partitionManager()
+ .createPartitions(
+ Collections.singletonList(repair.spec()),
+ true,
+ statistics,
+ true,
+ Collections.singletonList(repair.options()));
+ assertThat(registered.get(0).recordCount()).isOne();
+ assertThat(registered.get(0).fileCount()).isOne();
+ assertThat(registered.get(0).fileSizeInBytes())
+ .isEqualTo(table.fileIO().getFileStatus(replacement).getLen());
+ ReadBuilder read = table.newReadBuilder();
+ assertThat(readRows(read,
read.newScan().plan().splits())).containsExactly("replacement");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testOnlyOverwriteReportsMatchingFormat(boolean overwrite) throws
Exception {
+ Path location = new Path(tempDir.toUri());
+ List<Partition> existing =
+ Arrays.asList(
+ partition("inherited", null), partition("explicit",
options("parquet")));
+ FormatTable table = table(location, "parquet", existing);
+ AtomicReference<List<Map<String, String>>> reported = new
AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ reported.set(invocation.getArgument(4));
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ if (overwrite) {
+ builder.withOverwrite(Collections.emptyMap());
+ }
+ try (BatchTableWrite write = builder.newWrite()) {
+ for (String value : Arrays.asList("inherited", "explicit",
"created")) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString(value),
BinaryString.fromString(value)));
+ }
+ builder.newCommit().commit(write.prepareCommit());
+ }
+ if (overwrite) {
+ assertThat(reported.get())
+ .hasSize(3)
+ .allSatisfy(
+ options ->
assertThat(options).containsEntry("file.format", "parquet"));
+ } else {
+ assertThat(reported.get()).isNull();
+ }
+ List<Partition> registered = new ArrayList<>(existing);
+ registered.add(partition("created", null));
+ ReadBuilder read = table(location, "parquet",
registered).newReadBuilder();
+ assertThat(readRows(read, read.newScan().plan().splits()))
+ .containsExactlyInAnyOrder("inherited", "explicit", "created");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testOverwriteReportsFormatWhenRegistryChangesAfterLookup(boolean
explicitFormat)
+ throws Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "parquet", "old");
+ List<Partition> initial =
+ Collections.singletonList(
+ partition("p", explicitFormat ? options("parquet") :
null));
+ FormatTable table =
+ table(location, "parquet", initial)
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "false"));
+ AtomicReference<String> storedFormat =
+ new AtomicReference<>(explicitFormat ? "parquet" : null);
+ when(table.partitionManager().listPartitions(anyMap(), any()))
+ .thenAnswer(
+ invocation -> {
+ // The lookup returns its snapshot after another
request changes the
+ // registry. The overwrite must report what it
writes regardless.
+ storedFormat.set("orc");
+ return initial;
+ });
+ doAnswer(
+ invocation -> {
+ List<Map<String, String>> reportedOptions =
invocation.getArgument(4);
+ if (reportedOptions != null
+ &&
reportedOptions.get(0).containsKey("file.format")) {
+
storedFormat.set(reportedOptions.get(0).get("file.format"));
+ }
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+
+ BatchWriteBuilder overwrite =
+
table.newBatchWriteBuilder().withOverwrite(Collections.emptyMap());
+ try (BatchTableWrite write = overwrite.newWrite()) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("replacement"),
BinaryString.fromString("p")));
+ overwrite.newCommit().commit(write.prepareCommit());
+ }
+ assertThat(storedFormat.get()).isEqualTo("parquet");
+ ReadBuilder read =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(
+ partition("p",
options(storedFormat.get()))))
+ .newReadBuilder();
+ assertThat(readRows(read,
read.newScan().plan().splits())).containsExactly("replacement");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testTruncateDoesNotReportWriteFormat(boolean wholeTable) throws
Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "orc", "old");
+ FormatTable table =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(partition("p",
options("orc"))));
+ AtomicReference<List<Map<String, String>>> reported = new
AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ reported.set(invocation.getArgument(4));
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+ FormatTableCommit commit = (FormatTableCommit)
table.newBatchWriteBuilder().newCommit();
+ if (wholeTable) {
+ commit.truncateTable();
+ } else {
+ commit.truncatePartitions(
+ Collections.singletonList(Collections.singletonMap("pt",
"p")));
+ }
+ assertThat(reported.get())
+ .hasSize(1)
+ .allSatisfy(options ->
assertThat(options).doesNotContainKey("file.format"));
+ assertThat(FormatTableScan.listDataFiles(table.fileIO(), new
Path(location, "pt=p")))
+ .isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"static", "dynamic", "table"})
+ void testEmptyOverwriteUpdatesOnlyTargetedFormats(String overwriteMode)
throws Exception {
+ Path location = new Path(tempDir.toUri());
+ writeFile(location, "p", "orc", "old");
+ FormatTable table =
+ table(
+ location,
+ "parquet",
+ Collections.singletonList(partition("p",
options("orc"))))
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(),
+
Boolean.toString(overwriteMode.equals("dynamic"))));
+ AtomicReference<List<Map<String, String>>> reported = new
AtomicReference<>();
+ doAnswer(
+ invocation -> {
+ reported.set(invocation.getArgument(4));
+ return null;
+ })
+ .when(table.partitionManager())
+ .createPartitions(anyList(), anyBoolean(), any(),
anyBoolean(), any());
+ table.newBatchWriteBuilder()
+ .withOverwrite(
+ overwriteMode.equals("static")
+ ? Collections.singletonMap("pt", "p")
+ : Collections.emptyMap())
+ .newCommit()
+ .commit(Collections.emptyList());
+
+ if (overwriteMode.equals("dynamic")) {
+ assertThat(reported.get()).isNull();
+ ReadBuilder read = table.newReadBuilder();
+ assertThat(readRows(read,
read.newScan().plan().splits())).containsExactly("old");
+ } else {
+ assertThat(reported.get()).hasSize(1);
+ assertThat(reported.get().get(0)).containsEntry("file.format",
"parquet");
+ assertThat(FormatTableScan.listDataFiles(table.fileIO(), new
Path(location, "pt=p")))
+ .isEmpty();
+ }
+ }
+
+ private FormatTable table(Path location, String format, List<Partition>
partitions) {
+ FormatTablePartitionManager manager =
mock(FormatTablePartitionManager.class);
+ when(manager.listPartitions(anyMap(), any())).thenReturn(partitions);
+ when(manager.listPartitionsByNames(anyList())).thenReturn(partitions);
+ Map<String, String> tableOptions = options(format);
+ tableOptions.put(CoreOptions.PATH.key(), location.toString());
+ return FormatTable.builder()
+ .fileIO(LocalFileIO.create())
+ .identifier(Identifier.create("db", "mixed"))
+ .rowType(TABLE_TYPE)
+ .partitionKeys(Collections.singletonList("pt"))
+ .location(location.toString())
+ .format(FormatTable.parseFormat(format))
+ .catalogContext(CatalogContext.create(new Options()))
+ .options(tableOptions)
+ .partitionManager(manager)
+ .build();
+ }
+
+ private static Partition partition(String value, Map<String, String>
options) {
+ return new Partition(
+ Collections.singletonMap("pt", value),
+ 0,
+ 0,
+ 0,
+ 0,
+ -1,
+ false,
+ null,
+ null,
+ null,
+ null,
+ options);
+ }
+
+ private static Map<String, String> options(String format) {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.FILE_FORMAT.key(), format);
+ return options;
+ }
+
+ private static void writeFile(Path tablePath, String partition, String
format, String... values)
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tablePath, "pt=" + partition + "/data." + format);
+ fileIO.mkdirs(path.getParent());
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ FormatWriter writer =
+ FileFormat.fileFormat(new CoreOptions(options(format)))
+ .createWriterFactory(DATA_TYPE)
+ .create(out, "none");
+ for (String value : values) {
+
writer.addElement(GenericRow.of(BinaryString.fromString(value)));
+ }
+ writer.close();
+ }
+ }
+
+ private static List<String> readRows(ReadBuilder builder, List<Split>
splits) throws Exception {
+ List<String> rows = new ArrayList<>();
+ for (Split split : splits) {
+ try (RecordReader<InternalRow> reader =
+ builder.newRead().executeFilter().createReader(split)) {
+ reader.forEachRemaining(row ->
rows.add(row.getString(0).toString()));
+ }
+ }
+ return rows;
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
index 53669467f4..c272d98aa1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
@@ -707,6 +707,7 @@ public class FormatTableScanTest {
Map<String, String> options = new HashMap<>();
options.put(SOURCE_SPLIT_TARGET_SIZE.key(), "100b");
+ options.put(FILE_FORMAT.key(), format);
FormatTable formatTable =
createFormatTableWithOptions(
diff --git a/paimon-core/src/test/resources/compatibility/format-data-split-v3
b/paimon-core/src/test/resources/compatibility/format-data-split-v3
new file mode 100644
index 0000000000..5f3e8272f4
Binary files /dev/null and
b/paimon-core/src/test/resources/compatibility/format-data-split-v3 differ
diff --git a/paimon-flink/paimon-flink-common/pom.xml
b/paimon-flink/paimon-flink-common/pom.xml
index f0a38d8417..4478cb3a45 100644
--- a/paimon-flink/paimon-flink-common/pom.xml
+++ b/paimon-flink/paimon-flink-common/pom.xml
@@ -38,6 +38,22 @@ under the License.
</properties>
<dependencies>
+ <!-- Keep Paimon's format classes ahead of Flink SQL Gateway's ORC
dependencies
+ in reactor builds. Published connectors include them through
paimon-bundle. -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <!-- Use the Hadoop implementation supplied by this module's
parent. -->
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<dependency>
<groupId>org.apache.paimon</groupId>
<artifactId>${paimon-flinkx-common}</artifactId>
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
index ba8dc66aea..8a75b86f27 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FormatTableITCase.java
@@ -18,12 +18,20 @@
package org.apache.paimon.flink.source;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.RESTCatalogITCaseBase;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.rest.RESTToken;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
@@ -32,6 +40,10 @@ import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -39,6 +51,79 @@ import static org.assertj.core.api.Assertions.assertThat;
/** ITCase for format table. */
public class FormatTableITCase extends RESTCatalogITCaseBase {
+ @Test
+ public void testPartitionFileFormats() throws Exception {
+ String tableName = "partition_file_formats";
+ sql(
+ "CREATE TABLE %s (id INT, pt STRING) PARTITIONED BY (pt) WITH "
+ +
"('type'='format-table','file.format'='parquet','metastore.partitioned-table'='true')",
+ tableName);
+ setDataToken(tableName);
+ FormatTable table =
+ (FormatTable)
+
flinkCatalog().catalog().getTable(Identifier.create("default", tableName));
+ String[] formats = {"orc", "parquet"};
+ String[] values = {"old", "new"};
+ for (int i = 0; i < formats.length; i++) {
+ Path file = new Path(table.location(), "pt=" + values[i] +
"/data." + formats[i]);
+ table.fileIO().mkdirs(file.getParent());
+ try (PositionOutputStream out =
table.fileIO().newOutputStream(file, false)) {
+ FormatWriter writer =
+ FileFormat.fileFormat(
+ new CoreOptions(
+ Collections.singletonMap(
+ "file.format",
formats[i])))
+ .createWriterFactory(
+ RowType.builder().field("id",
DataTypes.INT()).build())
+ .create(out, "none");
+ writer.addElement(GenericRow.of(i + 1));
+ writer.close();
+ }
+ }
+ List<Map<String, String>> specs =
+ Arrays.asList(
+ Collections.singletonMap("pt", "old"),
+ Collections.singletonMap("pt", "new"));
+ table.partitionManager()
+ .createPartitions(
+ specs,
+ false,
+ null,
+ false,
+ Arrays.asList(
+ Collections.singletonMap("file.format", "orc"),
+ Collections.emptyMap()));
+ assertThat(sql("SELECT id, pt FROM %s", tableName))
+ .containsExactlyInAnyOrder(Row.of(1, "old"), Row.of(2, "new"));
+ assertThat(sql("SELECT id FROM %s WHERE pt='old' AND id=1", tableName))
+ .containsExactly(Row.of(1));
+ sql("INSERT INTO %s PARTITION (pt='new') VALUES (4)", tableName);
+ sql("INSERT OVERWRITE %s PARTITION (pt='new') VALUES (5)", tableName);
+ sql("INSERT INTO %s PARTITION (pt='created') VALUES (6)", tableName);
+ assertThat(
+ table.partitionManager()
+ .listPartitionsByNames(
+ Arrays.asList(
+ specs.get(1),
+ Collections.singletonMap("pt",
"created"))))
+ .satisfiesExactly(
+ overwritten ->
+ assertThat(overwritten.options())
+ .containsEntry("file.format",
"parquet"),
+ appended ->
assertThat(appended.options()).isNullOrEmpty());
+ sql("INSERT OVERWRITE %s PARTITION (pt='old') VALUES (3)", tableName);
+ assertThat(sql("SELECT id, pt FROM %s", tableName))
+ .containsExactlyInAnyOrder(
+ Row.of(3, "old"), Row.of(5, "new"), Row.of(6,
"created"));
+ assertThat(
+ table.partitionManager()
+
.listPartitionsByNames(Collections.singletonList(specs.get(0)))
+ .get(0)
+ .options())
+ .containsEntry("file.format", "parquet");
+ sql("DROP TABLE %s", tableName);
+ }
+
@Test
public void testDiffFormat() {
String bigDecimalStr = "10.001";
diff --git
a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java
b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java
index 6891ba9ecf..db88d765d8 100644
---
a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java
+++
b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java
@@ -138,6 +138,37 @@ class FormatTableMosaicReadTest {
assertThat(splits).isEmpty();
}
+ @Test
+ void testMosaicSplitOverridesTableFormat() throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("a", DataTypes.STRING())
+ .field("dt", DataTypes.STRING())
+ .build();
+ FormatTable table =
+ buildFormatTable(rowType, Collections.singletonList("dt"), new
HashMap<>());
+ writeAll(
+ table,
+ Collections.singletonList(
+ GenericRow.of(
+ BinaryString.fromString("value"),
BinaryString.fromString("p"))));
+ List<Split> splits = table.newReadBuilder().newScan().plan().splits();
+ FormatTable readerTable =
table.copy(Collections.singletonMap("file.format", "parquet"));
+ List<String> values = new ArrayList<>();
+ for (Split split : splits) {
+ try (RecordReader<InternalRow> reader =
+
readerTable.newReadBuilder().newRead().createReader(split)) {
+ reader.forEachRemaining(
+ row ->
+ values.add(
+ row.getString(0).toString()
+ + "/"
+ +
row.getString(1).toString()));
+ }
+ }
+ assertThat(values).containsExactly("value/p");
+ }
+
@Test
void testInvalidCompressionPropagatedFromWriter() {
RowType rowType = RowType.builder().field("a",
DataTypes.STRING()).build();
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
index d593821127..24b06d6309 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions
import org.apache.paimon.fs.Path
-import org.apache.paimon.partition.PartitionStatistics
+import org.apache.paimon.partition.{Partition, PartitionStatistics}
import org.apache.paimon.spark.catalyst.analysis.PaimonResolvePartitionSpec
import org.apache.paimon.spark.format.PaimonFormatTable
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
@@ -34,7 +34,7 @@ import
org.apache.spark.sql.catalyst.expressions.GenericInternalRow
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.unsafe.types.UTF8String
-import java.util.{Collections, List => JList, Map => JMap, Objects}
+import java.util.{Collections, List => JList, Objects}
import scala.collection.JavaConverters._
import scala.collection.immutable.ListMap
@@ -107,9 +107,9 @@ case class PaimonAnalyzeFormatTablePartitionsCommand(
if (noScan) {
// One listing request per partition is all NOSCAN reports, cheap
enough to take here.
new FormatTablePartitionStatsCollector(v2Table.table, false,
parallelism)
- .collect(partitions.asJava)
+ .collectPartitions(registeredPartitions.asJava)
} else {
- measureOnExecutors(sparkSession, partitions, parallelism)
+ measureOnExecutors(sparkSession, registeredPartitions, parallelism)
}
v2Table.partitionManager
.createPartitions(partitions.asJava, true, statistics, true, null)
@@ -128,7 +128,7 @@ case class PaimonAnalyzeFormatTablePartitionsCommand(
*/
private def measureOnExecutors(
sparkSession: SparkSession,
- partitions: List[JMap[String, String]],
+ partitions: List[Partition],
parallelism: Int): JList[PartitionStatistics] = {
val effectiveParallelism = math.max(1, parallelism)
val tasks = math.min(effectiveParallelism, partitions.size)
@@ -147,7 +147,7 @@ case class PaimonAnalyzeFormatTablePartitionsCommand(
work
.map(_._2)
.iterator
- .zip(collector.collect(work.map(_._1).asJava).asScala.iterator)
+
.zip(collector.collectPartitions(work.map(_._1).asJava).asScala.iterator)
}
}
.collect()
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
index b0430f853c..7ec61aa024 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
@@ -878,7 +878,13 @@ class FormatTablePartitionRepairTest {
@Override
public List<Partition> listPartitionsByNames(List<Map<String, String>>
partitions) {
- throw new UnsupportedOperationException();
+ List<Partition> result = new ArrayList<>(partitions.size());
+ for (Map<String, String> spec : partitions) {
+ if (registered.contains(spec)) {
+ result.add(partition(spec));
+ }
+ }
+ return result;
}
@Override
@@ -887,22 +893,25 @@ class FormatTablePartitionRepairTest {
requestedPrefixes.add(prefix);
List<Partition> partitions = new ArrayList<>(registered.size());
for (Map<String, String> spec : registered) {
- partitions.add(
- new Partition(
- spec,
- 0L,
- 0L,
- 0L,
- 0L,
- 0,
- false,
- null,
- null,
- null,
- null,
- partitionOptions.get(spec)));
+ partitions.add(partition(spec));
}
return partitions;
}
+
+ private Partition partition(Map<String, String> spec) {
+ return new Partition(
+ spec,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0,
+ false,
+ null,
+ null,
+ null,
+ null,
+ partitionOptions.get(spec));
+ }
}
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
index 2668929b2f..57aa2baf7a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
@@ -28,8 +28,62 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.TableCapability
import org.apache.spark.sql.connector.catalog.TableCatalog
+import scala.collection.JavaConverters._
+
class PaimonFormatTableTest extends PaimonSparkTestWithRestCatalogBase {
+ test("PaimonFormatTable: partition file formats survive reads statistics and
overwrite") {
+ val tableName = "partition_file_formats"
+ withTable(tableName) {
+ sql(s"CREATE TABLE $tableName (id INT, pt STRING) USING parquet
PARTITIONED BY (pt) " +
+ "TBLPROPERTIES ('metastore.partitioned-table'='true',
'format-table.implementation'='paimon')")
+ val table =
+ paimonCatalog.getTable(Identifier.create("test_db",
tableName)).asInstanceOf[FormatTable]
+ spark
+ .sql("SELECT 1 AS id")
+ .write
+ .format("orc")
+ .save(new Path(table.location(), "pt=old").toString)
+ spark.sql("SELECT 2 AS id").write.parquet(new Path(table.location(),
"pt=new").toString)
+ val specs = Seq(Map("pt" -> "old").asJava, Map("pt" ->
"new").asJava).asJava
+ table
+ .partitionManager()
+ .createPartitions(
+ specs,
+ false,
+ null,
+ false,
+ Seq(Map("file.format" -> "orc").asJava, Map.empty[String,
String].asJava).asJava)
+ checkAnswer(sql(s"SELECT id, pt FROM $tableName"), Seq(Row(1, "old"),
Row(2, "new")))
+ checkAnswer(sql(s"SELECT id FROM $tableName WHERE pt='old' AND id=1"),
Seq(Row(1)))
+ sql(s"ANALYZE TABLE $tableName COMPUTE STATISTICS")
+ assert(
+
table.partitionManager().listPartitionsByNames(specs).asScala.forall(_.recordCount()
== 1))
+ sql(s"INSERT INTO $tableName PARTITION (pt='new') VALUES (4)")
+ sql(s"INSERT OVERWRITE $tableName PARTITION (pt='new') VALUES (5)")
+ sql(s"INSERT INTO $tableName PARTITION (pt='created') VALUES (6)")
+ val writtenSpecs = Seq(specs.get(1), Map("pt" ->
"created").asJava).asJava
+ val writtenPartitions =
+ table
+ .partitionManager()
+ .listPartitionsByNames(writtenSpecs)
+ assert(writtenPartitions.size() == 2)
+ assert(writtenPartitions.get(0).options().get("file.format") ==
"parquet")
+ assert(Option(writtenPartitions.get(1).options()).forall(_.isEmpty))
+ sql(s"INSERT OVERWRITE $tableName PARTITION (pt='old') VALUES (3)")
+ checkAnswer(
+ sql(s"SELECT id, pt FROM $tableName"),
+ Seq(Row(3, "old"), Row(5, "new"), Row(6, "created")))
+ assert(
+ table
+ .partitionManager()
+ .listPartitionsByNames(Seq(specs.get(0)).asJava)
+ .get(0)
+ .options()
+ .get("file.format") == "parquet")
+ }
+ }
+
override protected def beforeEach(): Unit = {
super.beforeEach()
sql("USE paimon")
