[core] Support unknown row counts for bucket-unaware append tables
diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 6f72db9..5e8a19e 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md
@@ -350,7 +350,7 @@ | --- | --- | --- | | `_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 @@ [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 8c285c3..e32bf4d 100644 --- a/docs/docs/concepts/spec/snapshot.md +++ b/docs/docs/concepts/spec/snapshot.md
@@ -114,6 +114,16 @@ 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 d9de932..4126d6d 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 @@ @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 @@ 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 b880d5c..ce010d3 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
@@ -50,6 +50,12 @@ } @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, Snapshot snapshot,
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 ff912e2..5cc4e6b 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 @@ /** 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 48aaf07..ef70290 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 @@ 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 @@ 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 9d88923..6ebcf39 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 @@ 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 3e93bc8..6acb932 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ } BucketEntry that = (BucketEntry) o; return recordCount == that.recordCount + && unknownRowCount == that.unknownRowCount && fileSizeInBytes == that.fileSizeInBytes && fileCount == that.fileCount && lastFileCreationTime == that.lastFileCreationTime @@ -138,6 +164,12 @@ @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 d162234..6a92964 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 @@ 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 @@ 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 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 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 @@ fileSizeInBytes, fileCount, file.creationTimeEpochMillis(), - totalBuckets); + totalBuckets, + unknownRowCount); } public static Collection<PartitionEntry> merge(Collection<ManifestEntry> fileEntries) { @@ -163,6 +190,7 @@ } PartitionEntry that = (PartitionEntry) o; return recordCount == that.recordCount + && unknownRowCount == that.unknownRowCount && fileSizeInBytes == that.fileSizeInBytes && fileCount == that.fileCount && lastFileCreationTime == that.lastFileCreationTime @@ -175,6 +203,7 @@ 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 af38f1d..5f6e671 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 @@ 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 @@ /** 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 e518c8f..5b550e2 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.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 @@ } } + validateUnknownRowCounts( + deltaFiles, latestSnapshot != null && latestSnapshot.totalRecordCount() < 0); if (latestSnapshot == null) { conflictDetection.checkSameBucketWithinDelta(deltaFiles); } @@ -1217,8 +1220,13 @@ } // 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 @@ @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 @@ } } + 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 @@ 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 @@ null, latestSnapshot.watermark(), latestSnapshot.statistics(), - latestSnapshot.properties(), + recordCountProperties(latestSnapshot.properties(), emptyList()), latestSnapshot.nextRowId(), null); @@ -1730,6 +1740,7 @@ 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 @@ } } + 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 d2ed729..37bd13b 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 @@ 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 @@ 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 87b244d..e8cd1bd 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 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 @@ @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 78dbfff..ea3ac59 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 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 c34bdec..68edee5 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 @@ 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 bd7f781..1d12bce 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
@@ -108,6 +108,22 @@ } @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(); deletionVector.delete(0);
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 0000000..ac373cc --- /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 c135755..65326a8 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 @@ 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 @@ 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 0000000..e9ff56d --- /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 27f98e7..1c1819e 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 @@ 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 0000000..1b17454 --- /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() + } + } +}