You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/08 04:43:01 UTC
(paimon) 01/01: [core] Support unknown row counts for bucket-unaware append tables
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch codex/append-unknown-row-count
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f006a392cfa52c463eee6e74fa4f8d31f7d3def8
Author: JingsongLi <ji...@gmail.com>
AuthorDate: Thu Oct 8 12:41:15 2026 +0800
[core] Support unknown row counts for bucket-unaware append tables
---
docs/docs/concepts/spec/manifest.md | 16 +-
docs/docs/concepts/spec/snapshot.md | 10 +
.../src/main/java/org/apache/paimon/Snapshot.java | 8 +
.../paimon/catalog/RenamingSnapshotCommit.java | 6 +
.../org/apache/paimon/catalog/SnapshotCommit.java | 5 +
.../java/org/apache/paimon/io/DataFileMeta.java | 2 +
.../org/apache/paimon/io/FileIndexEvaluator.java | 4 +-
.../org/apache/paimon/manifest/BucketEntry.java | 40 ++-
.../org/apache/paimon/manifest/PartitionEntry.java | 37 +-
.../paimon/operation/AppendOnlyFileStoreScan.java | 9 +-
.../paimon/operation/FileStoreCommitImpl.java | 64 +++-
.../operation/RemoveUnexistingManifests.java | 4 +
.../org/apache/paimon/table/source/DataSplit.java | 6 +
.../paimon/table/source/IncrementalSplit.java | 6 +
.../paimon/utils/PartitionStatisticsReporter.java | 6 +-
.../apache/paimon/io/FileIndexEvaluatorTest.java | 16 +
.../apache/paimon/table/UnknownRowCountTest.java | 390 +++++++++++++++++++++
.../paimon/flink/source/SplitWeightUtils.java | 7 +-
.../paimon/flink/source/SplitWeightUtilsTest.java | 77 ++++
.../commands/PaimonAnalyzeTableColumnCommand.scala | 4 +-
.../paimon/spark/sql/UnknownRowCountTest.scala | 118 +++++++
21 files changed, 815 insertions(+), 20 deletions(-)
diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md
index 6f72db9067..5e8a19e546 100644
--- a/docs/docs/concepts/spec/manifest.md
+++ b/docs/docs/concepts/spec/manifest.md
@@ -350,7 +350,7 @@ The `_FILE` record includes file identity, statistics, and optional feature meta
| --- | --- | --- |
| `_FILE_NAME` | STRING | Data or changelog file name. |
| `_FILE_SIZE` | BIGINT | File size in bytes. |
-| `_ROW_COUNT` | BIGINT | Physical record count, including row kinds that represent deletions. |
+| `_ROW_COUNT` | BIGINT | Physical record count, including row kinds that represent deletions; `-1` means unknown for supported bucket-unaware append tables. |
| `_MIN_KEY`, `_MAX_KEY` | BYTES | Serialized BinaryRow key bounds, not SQL strings. |
| `_KEY_STATS`, `_VALUE_STATS` | SimpleStats | Statistics for key and value fields. |
| `_MIN_SEQUENCE_NUMBER`, `_MAX_SEQUENCE_NUMBER` | BIGINT | Sequence-number bounds. |
@@ -371,6 +371,20 @@ These counts describe physical files; they need not equal the logical table row
[Snapshot record counts](./snapshot#interpreting-record-counts) and
[Data Evolution](../../multimodal-table/data-evolution) for examples.
+An ordinary bucket-unaware append file may record `_ROW_COUNT = -1` when its row count is
+unavailable. Readers must preserve this uncertainty when combining files and must read records
+instead of using metadata-only count aggregation or position-based limit selections. Column
+statistics can be absent independently of the row count.
+
+Unknown file row counts currently require a renaming-based snapshot backend, with row tracking,
+data evolution and Iceberg metadata synchronization disabled. Catalog-managed snapshot commits
+do not opt in because their partition-statistics protocol interprets every negative record-count
+delta as a decrement. A known delta of `-1` still means deleting one physical record.
+
+Upgrade all readers and writers before producing unknown-count metadata. The binary manifest
+schema is unchanged, but older readers can deserialize `-1` and incorrectly use it as an exact
+count. This is a reader-semantics change, not compatibility with older readers.
+
### Index Manifest
An index manifest describes index files through these fields:
diff --git a/docs/docs/concepts/spec/snapshot.md b/docs/docs/concepts/spec/snapshot.md
index 8c285c3e79..e32bf4d732 100644
--- a/docs/docs/concepts/spec/snapshot.md
+++ b/docs/docs/concepts/spec/snapshot.md
@@ -114,6 +114,16 @@ Dedicated Format table stores the regular columns and each dedicated BLOB column
files. Appending `N` logical rows to a table with one dedicated BLOB column therefore increases the
`deltaRecordCount` by `2 * N`. Use `COUNT(*)` when you need the logical row count.
+For supported append tables with unknown file row counts, `totalRecordCount = -1` means the total
+is unknown. Ordinary commits preserve this conservative total after removing unknown files;
+they do not rescan every live manifest to recover an exact count. Metadata repair can recompute
+the total from surviving known file counts, and rollback can inherit an exact target total.
+
+`deltaRecordCount` is signed: `-1` normally means one record was deleted. If a commit adds or
+removes an unknown-count file, it sets `properties['delta-record-count.unknown'] = 'true'` and
+records an unknown delta instead. Use `Snapshot.deltaRecordCountKnown()` to distinguish those
+cases; the numeric value alone is insufficient. Subsequent known deltas clear the marker.
+
## Inspect and Retain Snapshots
Use the [snapshots system table](../system-tables#snapshots-table) to inspect commit history with
diff --git a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
index d9de932506..4126d6dbe2 100644
--- a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
@@ -43,6 +43,8 @@ import java.util.UUID;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Snapshot implements Serializable {
+ public static final String DELTA_RECORD_COUNT_UNKNOWN = "delta-record-count.unknown";
+
private static final long serialVersionUID = 1L;
public static final long FIRST_SNAPSHOT_ID = 1;
@@ -403,6 +405,12 @@ public class Snapshot implements Serializable {
return deltaRecordCount;
}
+ /** A signed delta of -1 can be known; unknown deltas are marked explicitly in properties. */
+ public boolean deltaRecordCountKnown() {
+ return properties == null
+ || !Boolean.parseBoolean(properties.get(DELTA_RECORD_COUNT_UNKNOWN));
+ }
+
@JsonGetter(FIELD_CHANGELOG_RECORD_COUNT)
@Nullable
public Long changelogRecordCount() {
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
index b880d5c4a2..ce010d367c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
@@ -49,6 +49,12 @@ public class RenamingSnapshotCommit implements SnapshotCommit {
this.lock = lock;
}
+ @Override
+ public boolean supportsUnknownRowCount() {
+ // Partition deltas are not sent to a catalog server by this backend.
+ return true;
+ }
+
@Override
public boolean commit(
@Nullable String baseSnapshotUuid,
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
index ff912e2139..5cc4e6b2db 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
@@ -28,6 +28,11 @@ import java.util.List;
/** Interface to commit snapshot atomically. */
public interface SnapshotCommit extends AutoCloseable {
+ /** Whether this backend can commit unknown file row counts without treating them as deltas. */
+ default boolean supportsUnknownRowCount() {
+ return false;
+ }
+
boolean commit(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
index 48aaf07809..ef702902e0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
@@ -119,6 +119,7 @@ public interface DataFileMeta {
BinaryRow EMPTY_MIN_KEY = EMPTY_ROW;
BinaryRow EMPTY_MAX_KEY = EMPTY_ROW;
int DUMMY_LEVEL = 0;
+ long UNKNOWN_ROW_COUNT = -1;
static DataFileMeta forAppend(
String fileName,
@@ -295,6 +296,7 @@ public interface DataFileMeta {
long fileSize();
+ /** Returns the file row count, or {@link #UNKNOWN_ROW_COUNT} when it is unavailable. */
long rowCount();
Optional<Long> deleteRowCount();
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
index 9d88923f2f..6ebcf396ee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
@@ -77,7 +77,9 @@ public class FileIndexEvaluator {
throws IOException {
// File index selections use 32-bit positions. Fall back when they cannot safely represent
// the file or its deletion vector.
- if (file.rowCount() > RoaringBitmap32.MAX_VALUE || dv instanceof Bitmap64DeletionVector) {
+ if (file.rowCount() < 0
+ || file.rowCount() > RoaringBitmap32.MAX_VALUE
+ || dv instanceof Bitmap64DeletionVector) {
return FileIndexResult.REMAIN;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java
index 3e93bc8cc1..6acb932260 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketEntry.java
@@ -20,6 +20,7 @@ package org.apache.paimon.manifest;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.utils.Pair;
import java.util.Collection;
@@ -34,6 +35,7 @@ public class BucketEntry {
private final BinaryRow partition;
private final int bucket;
private final long recordCount;
+ private final boolean unknownRowCount;
private final long fileSizeInBytes;
private final long fileCount;
private final long lastFileCreationTime;
@@ -45,9 +47,28 @@ public class BucketEntry {
long fileSizeInBytes,
long fileCount,
long lastFileCreationTime) {
+ this(
+ partition,
+ bucket,
+ recordCount,
+ fileSizeInBytes,
+ fileCount,
+ lastFileCreationTime,
+ false);
+ }
+
+ private BucketEntry(
+ BinaryRow partition,
+ int bucket,
+ long recordCount,
+ long fileSizeInBytes,
+ long fileCount,
+ long lastFileCreationTime,
+ boolean unknownRowCount) {
this.partition = partition;
this.bucket = bucket;
this.recordCount = recordCount;
+ this.unknownRowCount = unknownRowCount;
this.fileSizeInBytes = fileSizeInBytes;
this.fileCount = fileCount;
this.lastFileCreationTime = lastFileCreationTime;
@@ -81,10 +102,13 @@ public class BucketEntry {
return new BucketEntry(
partition,
bucket,
- recordCount + entry.recordCount,
+ unknownRowCount || entry.unknownRowCount
+ ? DataFileMeta.UNKNOWN_ROW_COUNT
+ : recordCount + entry.recordCount,
fileSizeInBytes + entry.fileSizeInBytes,
fileCount + entry.fileCount,
- Math.max(lastFileCreationTime, entry.lastFileCreationTime));
+ Math.max(lastFileCreationTime, entry.lastFileCreationTime),
+ unknownRowCount || entry.unknownRowCount);
}
public static BucketEntry fromManifestEntry(ManifestEntry entry) {
@@ -95,7 +119,8 @@ public class BucketEntry {
partitionEntry.recordCount(),
partitionEntry.fileSizeInBytes(),
partitionEntry.fileCount(),
- partitionEntry.lastFileCreationTime());
+ partitionEntry.lastFileCreationTime(),
+ entry.file().rowCount() < 0);
}
public static Collection<BucketEntry> merge(Collection<ManifestEntry> fileEntries) {
@@ -128,6 +153,7 @@ public class BucketEntry {
}
BucketEntry that = (BucketEntry) o;
return recordCount == that.recordCount
+ && unknownRowCount == that.unknownRowCount
&& fileSizeInBytes == that.fileSizeInBytes
&& fileCount == that.fileCount
&& lastFileCreationTime == that.lastFileCreationTime
@@ -138,6 +164,12 @@ public class BucketEntry {
@Override
public int hashCode() {
return Objects.hash(
- partition, bucket, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
+ partition,
+ bucket,
+ recordCount,
+ unknownRowCount,
+ fileSizeInBytes,
+ fileCount,
+ lastFileCreationTime);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionEntry.java
index d162234cd5..6a92964fc6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionEntry.java
@@ -38,6 +38,7 @@ public class PartitionEntry {
private final BinaryRow partition;
private final long recordCount;
+ private final boolean unknownRowCount;
private final long fileSizeInBytes;
private final long fileCount;
private final long lastFileCreationTime;
@@ -50,8 +51,27 @@ public class PartitionEntry {
long fileCount,
long lastFileCreationTime,
int totalBuckets) {
+ this(
+ partition,
+ recordCount,
+ fileSizeInBytes,
+ fileCount,
+ lastFileCreationTime,
+ totalBuckets,
+ false);
+ }
+
+ private PartitionEntry(
+ BinaryRow partition,
+ long recordCount,
+ long fileSizeInBytes,
+ long fileCount,
+ long lastFileCreationTime,
+ int totalBuckets,
+ boolean unknownRowCount) {
this.partition = partition;
this.recordCount = recordCount;
+ this.unknownRowCount = unknownRowCount;
this.fileSizeInBytes = fileSizeInBytes;
this.fileCount = fileCount;
this.lastFileCreationTime = lastFileCreationTime;
@@ -85,11 +105,14 @@ public class PartitionEntry {
public PartitionEntry merge(PartitionEntry entry) {
return new PartitionEntry(
partition,
- recordCount + entry.recordCount,
+ unknownRowCount || entry.unknownRowCount
+ ? DataFileMeta.UNKNOWN_ROW_COUNT
+ : recordCount + entry.recordCount,
fileSizeInBytes + entry.fileSizeInBytes,
fileCount + entry.fileCount,
Math.max(lastFileCreationTime, entry.lastFileCreationTime),
- entry.totalBuckets);
+ entry.totalBuckets,
+ unknownRowCount || entry.unknownRowCount);
}
public Partition toPartition(InternalRowPartitionComputer computer) {
@@ -120,10 +143,13 @@ public class PartitionEntry {
public static PartitionEntry fromDataFile(
BinaryRow partition, FileKind kind, DataFileMeta file, int totalBuckets) {
long recordCount = file.rowCount();
+ boolean unknownRowCount = recordCount < 0;
long fileSizeInBytes = file.fileSize();
long fileCount = 1;
if (kind == DELETE) {
- recordCount = -recordCount;
+ if (!unknownRowCount) {
+ recordCount = -recordCount;
+ }
fileSizeInBytes = -fileSizeInBytes;
fileCount = -fileCount;
}
@@ -133,7 +159,8 @@ public class PartitionEntry {
fileSizeInBytes,
fileCount,
file.creationTimeEpochMillis(),
- totalBuckets);
+ totalBuckets,
+ unknownRowCount);
}
public static Collection<PartitionEntry> merge(Collection<ManifestEntry> fileEntries) {
@@ -163,6 +190,7 @@ public class PartitionEntry {
}
PartitionEntry that = (PartitionEntry) o;
return recordCount == that.recordCount
+ && unknownRowCount == that.unknownRowCount
&& fileSizeInBytes == that.fileSizeInBytes
&& fileCount == that.fileCount
&& lastFileCreationTime == that.lastFileCreationTime
@@ -175,6 +203,7 @@ public class PartitionEntry {
return Objects.hash(
partition,
recordCount,
+ unknownRowCount,
fileSizeInBytes,
fileCount,
lastFileCreationTime,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
index af38f1d6c9..5f6e671447 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
@@ -115,7 +115,9 @@ public class AppendOnlyFileStoreScan extends AbstractFileStoreScan {
while (result.hasNext()) {
ManifestEntry next = result.next();
filtered.add(next);
- accumulatedRowCount += next.file().rowCount();
+ if (next.file().rowCount() >= 0) {
+ accumulatedRowCount += next.file().rowCount();
+ }
if (accumulatedRowCount >= limit) {
break;
}
@@ -126,6 +128,11 @@ public class AppendOnlyFileStoreScan extends AbstractFileStoreScan {
/** Note: Keep this thread-safe. */
@Override
protected boolean filterByStats(ManifestEntry entry) {
+ // Metadata-only imports have neither a row count nor column statistics. Schema evolution
+ // must not turn their unknown null counts into proof that a predicate cannot match.
+ if (entry.file().rowCount() < 0) {
+ return true;
+ }
Predicate notEvolvedFilter =
notEvolvedFilterMapping.computeIfAbsent(
entry.file().schemaId(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index e518c8f0bc..5b550e2c10 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -27,6 +27,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.iceberg.IcebergOptions;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
@@ -1053,6 +1054,8 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
}
+ validateUnknownRowCounts(
+ deltaFiles, latestSnapshot != null && latestSnapshot.totalRecordCount() < 0);
if (latestSnapshot == null) {
conflictDetection.checkSameBucketWithinDelta(deltaFiles);
}
@@ -1217,8 +1220,13 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
// the added records subtract the deleted records from
- long deltaRecordCount = recordCountAdd(deltaFiles) - recordCountDelete(deltaFiles);
- long totalRecordCount = previousTotalRecordCount + deltaRecordCount;
+ long deltaRecordCount = deltaRecordCount(deltaFiles);
+ boolean unknownRowCount = deltaFiles.stream().anyMatch(e -> e.file().rowCount() < 0);
+ properties = recordCountProperties(properties, deltaFiles);
+ long totalRecordCount =
+ unknownRowCount || previousTotalRecordCount < 0
+ ? DataFileMeta.UNKNOWN_ROW_COUNT
+ : previousTotalRecordCount + deltaRecordCount;
// write new delta files into manifest files
deltaPartitionEntries = new ArrayList<>(PartitionEntry.merge(deltaFiles));
@@ -1451,6 +1459,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
@Nullable String indexManifest,
@Nullable Long nextRowId,
@Nullable Map<String, String> properties) {
+ properties = recordCountProperties(properties, emptyList());
Snapshot newSnapshot =
new Snapshot(
latest.id() + 1,
@@ -1530,6 +1539,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
}
+ validateUnknownRowCounts(deltaFiles, targetSnapshot.totalRecordCount() < 0);
Pair<String, Long> baseManifestList =
manifestList.write(manifestFile.write(new ArrayList<>(latestEntries.values())));
Pair<String, Long> deltaManifestList = manifestList.write(manifestFile.write(deltaFiles));
@@ -1555,11 +1565,11 @@ public class FileStoreCommitImpl implements FileStoreCommit {
CommitKind.OVERWRITE,
System.currentTimeMillis(),
targetSnapshot.totalRecordCount(),
- recordCountAdd(deltaFiles) - recordCountDelete(deltaFiles),
+ deltaRecordCount(deltaFiles),
null,
targetSnapshot.watermark(),
targetSnapshot.statistics(),
- targetSnapshot.properties(),
+ recordCountProperties(targetSnapshot.properties(), deltaFiles),
nextRowId,
null);
@@ -1709,7 +1719,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
null,
latestSnapshot.watermark(),
latestSnapshot.statistics(),
- latestSnapshot.properties(),
+ recordCountProperties(latestSnapshot.properties(), emptyList()),
latestSnapshot.nextRowId(),
null);
@@ -1730,6 +1740,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
Snapshot newSnapshot,
List<PartitionEntry> deltaPartitionEntries) {
try {
+ validateUnknownRowCounts(emptyList(), newSnapshot.totalRecordCount() < 0);
List<PartitionStatistics> statistics = new ArrayList<>(deltaPartitionEntries.size());
for (PartitionEntry entry : deltaPartitionEntries) {
statistics.add(entry.toPartitionStatistics(partitionComputer));
@@ -1756,6 +1767,49 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
}
+ private void validateUnknownRowCounts(List<ManifestEntry> entries, boolean unknownTotal) {
+ if (!unknownTotal && entries.stream().noneMatch(e -> e.file().rowCount() < 0)) {
+ return;
+ }
+ checkArgument(
+ snapshotCommit.supportsUnknownRowCount(),
+ "Snapshot backend does not support unknown row counts.");
+ checkArgument(
+ bucketMode == BucketMode.BUCKET_UNAWARE
+ && !options.rowTrackingEnabled()
+ && !options.dataEvolutionEnabled(),
+ "Unknown row counts only support bucket-unaware append tables without row tracking.");
+ checkArgument(
+ options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
+ == IcebergOptions.StorageType.DISABLED,
+ "Unknown row counts do not support Iceberg metadata synchronization.");
+ }
+
+ private static long deltaRecordCount(List<ManifestEntry> entries) {
+ return entries.stream().anyMatch(e -> e.file().rowCount() < 0)
+ ? DataFileMeta.UNKNOWN_ROW_COUNT
+ : recordCountAdd(entries) - recordCountDelete(entries);
+ }
+
+ @Nullable
+ private static Map<String, String> recordCountProperties(
+ @Nullable Map<String, String> properties, List<ManifestEntry> entries) {
+ boolean unknown = entries.stream().anyMatch(e -> e.file().rowCount() < 0);
+ if (!unknown
+ && (properties == null
+ || !properties.containsKey(Snapshot.DELTA_RECORD_COUNT_UNKNOWN))) {
+ return properties;
+ }
+ Map<String, String> result =
+ properties == null ? new HashMap<>() : new HashMap<>(properties);
+ if (unknown) {
+ result.put(Snapshot.DELTA_RECORD_COUNT_UNKNOWN, "true");
+ } else {
+ result.remove(Snapshot.DELTA_RECORD_COUNT_UNKNOWN);
+ }
+ return result;
+ }
+
@Override
public void close() {
IOUtils.closeAllQuietly(commitPreCallbacks);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java b/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
index d2ed729581..37bd13b3d7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
@@ -21,6 +21,7 @@ package org.apache.paimon.operation;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
@@ -132,6 +133,9 @@ public class RemoveUnexistingManifests {
long total = 0L;
for (ManifestEntry entry : entries) {
if (entry.kind() == FileKind.ADD && !deleted.contains(entry.identifier())) {
+ if (entry.file().rowCount() < 0) {
+ return DataFileMeta.UNKNOWN_ROW_COUNT;
+ }
total += entry.file().rowCount();
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
index 87b244db57..e8cd1bd9f7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java
@@ -130,6 +130,9 @@ public class DataSplit implements Split {
public long rowCount() {
long rowCount = 0;
for (DataFileMeta file : dataFiles) {
+ if (file.rowCount() < 0) {
+ return DataFileMeta.UNKNOWN_ROW_COUNT;
+ }
rowCount += file.rowCount();
}
return rowCount;
@@ -137,6 +140,9 @@ public class DataSplit implements Split {
@Override
public OptionalLong mergedRowCount() {
+ if (dataFiles.stream().anyMatch(file -> file.rowCount() < 0)) {
+ return OptionalLong.empty();
+ }
if (rawMergedRowCountAvailable()) {
return OptionalLong.of(rawMergedRowCount());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java
index 78dbfff66f..ea3ac59c7a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/IncrementalSplit.java
@@ -127,9 +127,15 @@ public class IncrementalSplit implements Split {
public long rowCount() {
long rowCount = 0;
for (DataFileMeta file : beforeFiles) {
+ if (file.rowCount() < 0) {
+ return DataFileMeta.UNKNOWN_ROW_COUNT;
+ }
rowCount += file.rowCount();
}
for (DataFileMeta file : afterFiles) {
+ if (file.rowCount() < 0) {
+ return DataFileMeta.UNKNOWN_ROW_COUNT;
+ }
rowCount += file.rowCount();
}
return rowCount;
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionStatisticsReporter.java b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionStatisticsReporter.java
index c34bdec5b5..68edee52ef 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionStatisticsReporter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionStatisticsReporter.java
@@ -80,7 +80,11 @@ public class PartitionStatisticsReporter implements Closeable {
List<DataFileMeta> fileMetas = split.dataFiles();
fileCount += fileMetas.size();
for (DataFileMeta fileMeta : fileMetas) {
- rowCount += fileMeta.rowCount();
+ if (fileMeta.rowCount() < 0 || rowCount < 0) {
+ rowCount = DataFileMeta.UNKNOWN_ROW_COUNT;
+ } else {
+ rowCount += fileMeta.rowCount();
+ }
totalSize += fileMeta.fileSize();
}
totalBuckets = Optional.ofNullable(split.totalBuckets()).orElse(0);
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
index bd7f781855..1d12bce0cc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
@@ -107,6 +107,22 @@ public class FileIndexEvaluatorTest {
assertThat(result).isSameAs(FileIndexResult.REMAIN);
}
+ @Test
+ public void testUnknownRowCountDoesNotBuildLimitSelection() throws Exception {
+ FileIndexResult result =
+ FileIndexEvaluator.evaluate(
+ null,
+ null,
+ Collections.emptyList(),
+ null,
+ 10,
+ null,
+ fileWithRowCount(DataFileMeta.UNKNOWN_ROW_COUNT),
+ null);
+
+ assertThat(result).isSameAs(FileIndexResult.REMAIN);
+ }
+
@Test
public void testDataFilterIntersectsDeletionVector() throws Exception {
BitmapDeletionVector deletionVector = new BitmapDeletionVector();
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/UnknownRowCountTest.java b/paimon-core/src/test/java/org/apache/paimon/table/UnknownRowCountTest.java
new file mode 100644
index 0000000000..ac373cc202
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/UnknownRowCountTest.java
@@ -0,0 +1,390 @@
+/*
+ * 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;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.RenamingSnapshotCommit;
+import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.operation.RemoveUnexistingManifests;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.sink.BatchTableCommit;
+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.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.IncrementalSplit;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.PartitionStatisticsReporter;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.data.BinaryString.fromString;
+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.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+/** Tests unknown file row counts without a file-import producer. */
+public class UnknownRowCountTest extends TableTestBase {
+
+ @Override
+ protected Schema schemaDefault() {
+ return Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .option("bucket", "-1")
+ .option("file.format", "parquet")
+ .build();
+ }
+
+ @Test
+ public void testUnknownCountSurvivesSerializationAndMixedFiles() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, true, 1);
+ Snapshot first = table.snapshotManager().latestSnapshot();
+ assertThat(first.deltaRecordCountKnown()).isFalse();
+ assertThat(Snapshot.fromJson(first.toJson()).deltaRecordCountKnown()).isFalse();
+ commit(table, false, 2, 3);
+ DataSplit split = (DataSplit) table.newScan().plan().splits().get(0);
+ assertThat(split.rowCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(split.mergedRowCount()).isEmpty();
+ DataFileMeta unknown =
+ split.dataFiles().stream().filter(f -> f.rowCount() < 0).findFirst().get();
+ DataFileMetaSerializer serializer = new DataFileMetaSerializer();
+ assertThat(serializer.fromRow(serializer.toRow(unknown)).rowCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(snapshot.totalRecordCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(snapshot.deltaRecordCount()).isEqualTo(2);
+ assertThat(snapshot.deltaRecordCountKnown()).isTrue();
+ assertThat(table.newScan().listPartitionEntries().get(0).recordCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ List<ManifestEntry> entries =
+ split.dataFiles().stream()
+ .map(f -> ManifestEntry.create(FileKind.ADD, split.partition(), 0, -1, f))
+ .collect(Collectors.toList());
+ assertThat(BucketEntry.merge(entries).iterator().next().recordCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(
+ new IncrementalSplit(
+ 1,
+ split.partition(),
+ 0,
+ -1,
+ Arrays.asList(unknown),
+ null,
+ split.dataFiles(),
+ null,
+ false)
+ .rowCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(read(table).stream().map(row -> row.getInt(0)))
+ .containsExactlyInAnyOrder(1, 2, 3);
+ }
+
+ @Test
+ public void testKnownDeleteOfOneRowIsNotUnknown() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, false, 1);
+ DataSplit split = (DataSplit) table.newScan().plan().splits().get(0);
+ DataFileMeta file = split.dataFiles().get(0);
+ ManifestEntry add = ManifestEntry.create(FileKind.ADD, split.partition(), 0, -1, file);
+ ManifestEntry delete =
+ ManifestEntry.create(FileKind.DELETE, split.partition(), 0, -1, file);
+ assertThat(
+ PartitionEntry.fromManifestEntry(delete)
+ .merge(PartitionEntry.fromManifestEntry(add))
+ .recordCount())
+ .isZero();
+ assertThat(
+ BucketEntry.fromManifestEntry(delete)
+ .merge(BucketEntry.fromManifestEntry(add))
+ .recordCount())
+ .isZero();
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ Collections.singletonList(Collections.singletonMap("dt", "p1")));
+ }
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ assertThat(latest.deltaRecordCount()).isEqualTo(-1);
+ assertThat(latest.deltaRecordCountKnown()).isTrue();
+ assertThat(latest.totalRecordCount()).isZero();
+ assertThat(table.newScan().plan().splits()).isEmpty();
+ }
+
+ @Test
+ public void testUnknownCountsDoNotPruneNewNullableColumn() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, true, 1, 2, 3);
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(SchemaChange.addColumn("added", DataTypes.INT())),
+ false);
+ table = getTableDefault();
+ org.apache.paimon.predicate.PredicateBuilder predicates =
+ new org.apache.paimon.predicate.PredicateBuilder(table.rowType());
+ ReadBuilder builder = table.newReadBuilder().withFilter(predicates.isNull(2));
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ builder.newRead().createReader(builder.newScan().plan())) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactlyInAnyOrder(1, 2, 3);
+ assertThat(table.newReadBuilder().withLimit(2).newScan().plan().splits()).isNotEmpty();
+ }
+
+ @Test
+ public void testPartitionReporterKeepsUnknownAndOtherMeasurements() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, true, 1);
+ commit(table, false, 2, 3, 4, 5, 6);
+ PartitionModification modification = mock(PartitionModification.class);
+ try (PartitionStatisticsReporter reporter =
+ new PartitionStatisticsReporter(table, modification)) {
+ reporter.report("dt=p1/", 1000);
+ }
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<PartitionStatistics>> captor = ArgumentCaptor.forClass(List.class);
+ verify(modification).alterPartitions(captor.capture());
+ PartitionStatistics stats = captor.getValue().get(0);
+ assertThat(stats.recordCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(stats.fileCount()).isEqualTo(2);
+ assertThat(stats.fileSizeInBytes()).isGreaterThan(0);
+ assertThat(stats.lastFileCreationTime()).isEqualTo(1000);
+ }
+
+ @Test
+ public void testSnapshotTotalRemainsConservativelyUnknownAfterTruncate() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, true, 1);
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ Collections.singletonList(Collections.singletonMap("dt", "p1")));
+ }
+ assertThat(table.newScan().plan().splits()).isEmpty();
+ assertThat(table.snapshotManager().latestSnapshot().totalRecordCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ }
+
+ @Test
+ public void testBackendRejectsUnknownBeforePublicationButAllowsKnownDelete() throws Exception {
+ createTableDefault();
+ FileStoreTable original = getTableDefault();
+ SnapshotCommit backend =
+ spy(new RenamingSnapshotCommit(original.snapshotManager(), Lock.empty()));
+ doReturn(false).when(backend).supportsUnknownRowCount();
+ CatalogEnvironment environment = spy(CatalogEnvironment.empty());
+ doReturn(backend).when(environment).snapshotCommit(any());
+ FileStoreTable table =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(), original.schema(), environment);
+ assertThatThrownBy(() -> commit(table, true, 1))
+ .hasStackTraceContaining("Snapshot backend does not support unknown row counts");
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ commit(table, false, 2);
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ Collections.singletonList(Collections.singletonMap("dt", "p1")));
+ }
+ assertThat(table.snapshotManager().latestSnapshot().totalRecordCount()).isZero();
+ assertThat(table.snapshotManager().latestSnapshot().deltaRecordCount()).isEqualTo(-1);
+ assertThat(table.snapshotManager().latestSnapshot().deltaRecordCountKnown()).isTrue();
+ }
+
+ @Test
+ public void testRepairDoesNotInventCountsForSurvivingUnknownFiles() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, false, 2, 3, 4, 5, 6);
+ commit(table, false, 7);
+ commit(table, true, 1);
+ List<ManifestFileMeta> manifests =
+ table.store()
+ .newScan()
+ .manifestsReader()
+ .read(table.snapshotManager().latestSnapshot(), ScanMode.ALL)
+ .allManifests;
+ ManifestFileMeta dropped =
+ manifests.stream()
+ .filter(
+ meta ->
+ table.store().newScan().readManifest(meta).stream()
+ .anyMatch(e -> e.file().rowCount() == 1))
+ .reduce((left, right) -> right)
+ .get();
+ table.fileIO()
+ .delete(table.store().pathFactory().toManifestFilePath(dropped.fileName()), false);
+ assertThat(new RemoveUnexistingManifests(table).execute()).isTrue();
+ assertThat(table.snapshotManager().latestSnapshot().totalRecordCount())
+ .isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(table.snapshotManager().latestSnapshot().deltaRecordCountKnown()).isTrue();
+ }
+
+ @Test
+ public void testManifestCompactionClearsUnknownDeltaMarker() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, false, 1);
+ commit(table, true, 2);
+ Snapshot before = table.snapshotManager().latestSnapshot();
+ assertThat(before.deltaRecordCountKnown()).isFalse();
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.compactManifests();
+ }
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ assertThat(latest.id()).isGreaterThan(before.id());
+ assertThat(latest.deltaRecordCount()).isZero();
+ assertThat(latest.deltaRecordCountKnown()).isTrue();
+ assertThat(latest.totalRecordCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ }
+
+ @Test
+ public void testEnablingIcebergSyncRejectsKnownAppendAfterUnknownSnapshot() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, true, 1);
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(
+ SchemaChange.setOption(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location")),
+ false);
+ FileStoreTable synced = getTableDefault();
+ assertThatThrownBy(() -> commit(synced, false, 2))
+ .hasStackTraceContaining(
+ "Unknown row counts do not support Iceberg metadata synchronization");
+ assertThat(synced.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ assertThat(read(synced).stream().map(row -> row.getInt(0))).containsExactly(1);
+ }
+
+ @Test
+ public void testRollbackPreservesKnownTotalAndUnknownDelta() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ commit(table, false, 1);
+ Snapshot target = table.snapshotManager().latestSnapshot();
+ commit(table, true, 2);
+ try (org.apache.paimon.operation.FileStoreCommitImpl commit =
+ (org.apache.paimon.operation.FileStoreCommitImpl)
+ table.store().newCommit("rollback", table)) {
+ assertThat(commit.rollbackToAsLatest(target)).isTrue();
+ }
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ assertThat(latest.totalRecordCount()).isEqualTo(1);
+ assertThat(latest.deltaRecordCount()).isEqualTo(DataFileMeta.UNKNOWN_ROW_COUNT);
+ assertThat(latest.deltaRecordCountKnown()).isFalse();
+ assertThat(read(table).stream().map(row -> row.getInt(0))).containsExactly(1);
+ }
+
+ private void commit(FileStoreTable table, boolean unknownCount, int... ids) throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int id : ids) {
+ write.write(GenericRow.of(id, fromString("p1")));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ if (unknownCount) {
+ messages =
+ messages.stream()
+ .map(
+ message -> {
+ CommitMessageImpl m = (CommitMessageImpl) message;
+ List<DataFileMeta> files =
+ m.newFilesIncrement().newFiles().stream()
+ .map(
+ file ->
+ DataFileMeta.forAppend(
+ file.fileName(),
+ file.fileSize(),
+ DataFileMeta
+ .UNKNOWN_ROW_COUNT,
+ SimpleStats
+ .EMPTY_STATS,
+ file
+ .minSequenceNumber(),
+ file
+ .maxSequenceNumber(),
+ file.schemaId(),
+ file
+ .extraFiles(),
+ null,
+ FileSource
+ .APPEND,
+ Collections
+ .emptyList(),
+ file.externalPath()
+ .orElse(
+ null),
+ null,
+ null))
+ .collect(Collectors.toList());
+ return new CommitMessageImpl(
+ m.partition(),
+ m.bucket(),
+ m.totalBuckets(),
+ new DataIncrement(
+ files,
+ Collections.emptyList(),
+ Collections.emptyList()),
+ m.compactIncrement());
+ })
+ .collect(Collectors.toList());
+ }
+ commit.commit(messages);
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SplitWeightUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SplitWeightUtils.java
index c1357552f2..65326a8434 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SplitWeightUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SplitWeightUtils.java
@@ -45,7 +45,10 @@ final class SplitWeightUtils {
case FILE_SIZE:
return SplitWeightUtils::splitFileSizeOrRowCount;
case ROW_COUNT:
- return split -> split.split().rowCount();
+ return split -> {
+ long rowCount = split.split().rowCount();
+ return rowCount < 0 ? Math.max(1, splitFileSizeOrRowCount(split)) : rowCount;
+ };
default:
throw new UnsupportedOperationException(
"Unsupported split weight mode "
@@ -77,6 +80,6 @@ final class SplitWeightUtils {
return ((DataSplit) split)
.dataFiles().stream().mapToLong(file -> file.fileSize()).sum();
}
- return split.rowCount();
+ return Math.max(1, split.rowCount());
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SplitWeightUtilsTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SplitWeightUtilsTest.java
new file mode 100644
index 0000000000..e9ff56df8f
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SplitWeightUtilsTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.flink.source;
+
+import org.apache.paimon.flink.FlinkConnectorOptions;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.IncrementalSplit;
+import org.apache.paimon.table.source.QueryAuthSplit;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for split assignment weights of metadata-only imports. */
+class SplitWeightUtilsTest {
+
+ @Test
+ void testUnknownIncrementalCountHasPositiveFileSizeFallback() {
+ IncrementalSplit split = mock(IncrementalSplit.class);
+ when(split.rowCount()).thenReturn(DataFileMeta.UNKNOWN_ROW_COUNT);
+ QueryAuthSplit wrapped = mock(QueryAuthSplit.class);
+ when(wrapped.split()).thenReturn(split);
+ Options options = new Options();
+ options.set(
+ FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_WEIGHT_MODE,
+ FlinkConnectorOptions.SplitWeightMode.FILE_SIZE);
+ assertThat(
+ SplitWeightUtils.splitWeightFunc(options)
+ .apply(new FileStoreSourceSplit("unknown", split)))
+ .isEqualTo(1L);
+ assertThat(
+ SplitWeightUtils.splitWeightFunc(options)
+ .apply(new FileStoreSourceSplit("wrapped", wrapped)))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void testUnknownRowCountUsesFileSizeForAssignment() {
+ DataFileMeta file = mock(DataFileMeta.class);
+ when(file.fileSize()).thenReturn(123L);
+ DataSplit split = mock(DataSplit.class);
+ when(split.rowCount()).thenReturn(DataFileMeta.UNKNOWN_ROW_COUNT);
+ when(split.dataFiles()).thenReturn(Collections.singletonList(file));
+
+ assertThat(
+ SplitWeightUtils.splitWeightFunc(new Options())
+ .apply(new FileStoreSourceSplit("unknown", split)))
+ .isEqualTo(123L);
+ when(split.rowCount()).thenReturn(3L);
+ assertThat(
+ SplitWeightUtils.splitWeightFunc(new Options())
+ .apply(new FileStoreSourceSplit("known", split)))
+ .isEqualTo(3L);
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
index 27f98e71b6..1c1819ed0f 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
@@ -68,7 +68,9 @@ case class PaimonAnalyzeTableColumnCommand(
val (mergedRecordCount, colStats) =
PaimonStatsUtils.computeColumnStats(sparkSession, relation, attributes)
- val totalRecordCount = currentSnapshot.totalRecordCount()
+ val snapshotRecordCount = currentSnapshot.totalRecordCount()
+ val totalRecordCount =
+ if (snapshotRecordCount < 0) mergedRecordCount else snapshotRecordCount
checkState(totalRecordCount >= mergedRecordCount)
val mergedRecordSize = (totalSize.toDouble * mergedRecordCount / totalRecordCount).toLong
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UnknownRowCountTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UnknownRowCountTest.scala
new file mode 100644
index 0000000000..1b17454a98
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UnknownRowCountTest.scala
@@ -0,0 +1,118 @@
+/*
+ * 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.spark.sql
+
+import org.apache.paimon.data.{BinaryString, GenericRow}
+import org.apache.paimon.io.{DataFileMeta, DataIncrement}
+import org.apache.paimon.manifest.FileSource
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.stats.SimpleStats
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
+
+import org.apache.spark.sql.Row
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+
+class UnknownRowCountTest extends PaimonSparkTestBase {
+
+ Seq("parquet", "orc", "avro").foreach {
+ format =>
+ test(s"query and analyze $format files with unknown row counts") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, dt STRING)
+ |PARTITIONED BY (dt)
+ |TBLPROPERTIES ('bucket'='-1', 'file.format'='$format')
+ |""".stripMargin)
+ spark.sql("INSERT INTO T VALUES (4, 'p1'), (5, 'p2')")
+ commitUnknownRows(loadTable("T"))
+ spark.catalog.refreshTable("T")
+
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(5L))
+ checkAnswer(
+ spark.sql("SELECT dt, COUNT(*) FROM T GROUP BY dt"),
+ Seq(Row("p1", 4L), Row("p2", 1L)))
+ checkAnswer(
+ spark.sql("SELECT id FROM T WHERE id > 2 ORDER BY id LIMIT 2"),
+ Seq(Row(3), Row(4)))
+ assert(spark.sql("SELECT * FROM T LIMIT 3").collect().length == 3)
+ spark.sql("ALTER TABLE T ADD COLUMN added INT")
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T WHERE added IS NULL"), Row(5L))
+ spark.sql("ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS id")
+ checkAnswer(spark.sql("SELECT COUNT(*) FROM T"), Row(5L))
+ }
+ }
+
+ private def commitUnknownRows(table: FileStoreTable): Unit = {
+ val builder = table.newBatchWriteBuilder()
+ val write = builder.newWrite()
+ val commit = builder.newCommit()
+ try {
+ (1 to 3).foreach(id => write.write(GenericRow.of(Int.box(id), BinaryString.fromString("p1"))))
+ val messages: java.util.List[CommitMessage] = write
+ .prepareCommit()
+ .asScala
+ .map {
+ message =>
+ val m = message.asInstanceOf[CommitMessageImpl]
+ val files = m
+ .newFilesIncrement()
+ .newFiles()
+ .asScala
+ .map {
+ file =>
+ DataFileMeta.forAppend(
+ file.fileName(),
+ file.fileSize(),
+ DataFileMeta.UNKNOWN_ROW_COUNT,
+ SimpleStats.EMPTY_STATS,
+ file.minSequenceNumber(),
+ file.maxSequenceNumber(),
+ file.schemaId(),
+ file.extraFiles(),
+ null,
+ FileSource.APPEND,
+ Collections.emptyList[String](),
+ file.externalPath().orElse(null),
+ null,
+ null
+ )
+ }
+ .asJava
+ new CommitMessageImpl(
+ m.partition(),
+ m.bucket(),
+ m.totalBuckets(),
+ new DataIncrement(
+ files,
+ Collections.emptyList[DataFileMeta](),
+ Collections.emptyList[DataFileMeta]()),
+ m.compactIncrement()
+ ): CommitMessage
+ }
+ .asJava
+ commit.commit(messages)
+ } finally {
+ write.close()
+ commit.close()
+ }
+ }
+}