From 20d45d4a91e3ca3c2d7a8ff40625e35370527f80 Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Wed, 21 Oct 2026 23:07:57 +0900 Subject: [PATCH] [mosaic] Drop float bounds that can hide NaN or -0.0 The native Mosaic writer updates FLOAT/DOUBLE min/max with a partial comparison: a leading NaN pins both bounds to NaN, a later NaN never enters the bounds, and -0.0 equals +0.0. Paimon compares bounds with compareTo, so manifest pruning and row-group pruning could skip files and row groups that still contain matching rows. Mosaic stats carry no NaN count, so such bounds cannot be detected and are no longer used. Null counts are kept. Generated-by: Claude Code --- docs/docs/concepts/spec/fileformat.md | 2 +- .../paimon/format/mosaic/MosaicObjects.java | 17 ++ .../format/mosaic/MosaicRecordsReader.java | 5 +- .../mosaic/MosaicSimpleStatsExtractor.java | 5 +- .../mosaic/MosaicFloatingPointBoundsTest.java | 172 ++++++++++++++++++ .../MosaicSimpleStatsExtractorTest.java | 13 ++ 6 files changed, 209 insertions(+), 5 deletions(-) create mode 100644 paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicFloatingPointBoundsTest.java diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 18c1bb59d969..68e1290bcc22 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -284,7 +284,7 @@ Format Options: | Option | Default | Type | Description | | --- | --- | --- | --- | | `mosaic.num-buckets` | auto | Integer | Number of column buckets for parallel I/O. When set to 0 or not specified, the format auto-determines the bucket count. | -| `mosaic.stats-columns` | (empty) | String | Comma-separated column names to collect min/max statistics for filter pushdown. Empty means no statistics are collected. | +| `mosaic.stats-columns` | (empty) | String | Comma-separated column names to collect min/max statistics for filter pushdown. FLOAT and DOUBLE columns only keep null counts, because their min/max can hide NaN or -0.0. Empty means no statistics are collected. | | `mosaic.read.prefetch-row-groups` | 8 | Integer | Number of row groups a reader opens ahead of the one being consumed. Opening a row group issues several dependent range reads, so prefetching overlaps that latency with decoding. Each row group ahead keeps its decoded batch in memory and uses its own input stream, see `mosaic.read.prefetch-max-bytes`. 0 disables prefetching. | | `mosaic.read.prefetch-max-bytes` | 64 mb | MemorySize | Upper bound on the estimated decoded size of the row groups a reader keeps ahead, from their row counts and the projected column types. Wide projections or large row groups therefore lower the effective `mosaic.read.prefetch-row-groups`. | diff --git a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicObjects.java b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicObjects.java index 54d15c43c09e..e087c729f5f6 100644 --- a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicObjects.java +++ b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicObjects.java @@ -21,7 +21,9 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Decimal; import org.apache.paimon.data.Timestamp; +import org.apache.paimon.mosaic.ColumnStatistics; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.DecimalType; import org.apache.paimon.types.LocalZonedTimestampType; import org.apache.paimon.types.TimestampType; @@ -86,6 +88,21 @@ public static Object convertStatsValue(byte[] bytes, DataType dataType) { } } + /** + * Mosaic updates FLOAT and DOUBLE min/max with a partial comparison. A leading NaN pins both + * bounds to NaN, a later NaN never replaces a bound, and {@code -0.0} never replaces {@code + * +0.0} (or the reverse). Any of those can make a predicate skip a row group or file that still + * contains a matching row, and Mosaic stats carry no NaN count to detect it. A missing bound is + * not used for skipping. + */ + static boolean hasUsableMinMax(ColumnStatistics stats, DataType dataType) { + if (!stats.hasMinMax()) { + return false; + } + DataTypeRoot root = dataType.getTypeRoot(); + return root != DataTypeRoot.FLOAT && root != DataTypeRoot.DOUBLE; + } + private static Timestamp convertTimestamp(ByteBuffer buf, int precision) { if (precision <= 3) { return Timestamp.fromEpochMillis(buf.getLong()); diff --git a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java index f0e1018b9073..d5869c54ad5c 100644 --- a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java +++ b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsReader.java @@ -65,6 +65,7 @@ import java.util.concurrent.FutureTask; import static org.apache.paimon.format.mosaic.MosaicObjects.convertStatsValue; +import static org.apache.paimon.format.mosaic.MosaicObjects.hasUsableMinMax; import static org.apache.paimon.utils.Preconditions.checkArgument; /** File reader for Mosaic format. */ @@ -533,8 +534,8 @@ private boolean matchesRowGroup(int rowGroupIndex, long rowCount) { } nullCounts[i] = stats.getNullCount(); - if (stats.hasMinMax()) { - DataType dataType = fields.get(i).type(); + DataType dataType = fields.get(i).type(); + if (hasUsableMinMax(stats, dataType)) { Object min = convertStatsValue(stats.getMin(), dataType); Object max = convertStatsValue(stats.getMax(), dataType); minValues.setField(i, min); diff --git a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java index f426b27cfa81..421682db884a 100644 --- a/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java +++ b/paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java @@ -41,6 +41,7 @@ import java.util.Set; import static org.apache.paimon.format.mosaic.MosaicObjects.convertStatsValue; +import static org.apache.paimon.format.mosaic.MosaicObjects.hasUsableMinMax; /** Extracts statistics from Mosaic file metadata. */ public class MosaicSimpleStatsExtractor implements SimpleStatsExtractor { @@ -120,8 +121,8 @@ private SimpleColStats[] extractFromStats( seenColumns.add(colIdx); nullCounts[colIdx] += stat.getNullCount(); - if (stat.hasMinMax()) { - DataType dataType = rowType.getFields().get(colIdx).type(); + DataType dataType = rowType.getFields().get(colIdx).type(); + if (hasUsableMinMax(stat, dataType)) { Object min = convertStatsValue(stat.getMin(), dataType); Object max = convertStatsValue(stat.getMax(), dataType); if (min instanceof Comparable) { diff --git a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicFloatingPointBoundsTest.java b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicFloatingPointBoundsTest.java new file mode 100644 index 000000000000..eff42a2211f1 --- /dev/null +++ b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicFloatingPointBoundsTest.java @@ -0,0 +1,172 @@ +/* + * 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.format.mosaic; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.Table; +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.source.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Table reads of mosaic files whose FLOAT/DOUBLE stats are collected. Native bounds can hide NaN or + * {@code -0.0}, so neither manifest pruning nor row-group pruning may drop a matching row. + */ +class MosaicFloatingPointBoundsTest { + + @TempDir java.nio.file.Path tempDir; + + @BeforeAll + static void checkNativeLibrary() { + assumeTrue(isNativeAvailable(), "Mosaic native library not available"); + } + + @ParameterizedTest + @ValueSource(strings = {"float", "double"}) + void testLeadingNaNDoesNotHideFiniteRows(String typeName) throws Exception { + boolean isFloat = "float".equals(typeName); + Table table = createTable(isFloat); + write(table, value(isFloat, Double.NaN), value(isFloat, -5.0), value(isFloat, 3.0)); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + assertMatches(table, builder.lessThan(1, value(isFloat, 0.0)), 2); + assertMatches(table, builder.equal(1, value(isFloat, -5.0)), 2); + } + + @ParameterizedTest + @ValueSource(strings = {"float", "double"}) + void testSignedZeroIsNotHidden(String typeName) throws Exception { + boolean isFloat = "float".equals(typeName); + Table positiveFirst = createTable(isFloat, "positive_first"); + write(positiveFirst, value(isFloat, 0.0), value(isFloat, -0.0)); + Table negativeFirst = createTable(isFloat, "negative_first"); + write(negativeFirst, value(isFloat, -0.0), value(isFloat, 0.0)); + + PredicateBuilder builder = new PredicateBuilder(positiveFirst.rowType()); + assertMatches(positiveFirst, builder.equal(1, value(isFloat, -0.0)), 2); + assertMatches(negativeFirst, builder.equal(1, value(isFloat, 0.0)), 2); + } + + /** + * Checks the manifest level (the filtered scan still plans the file), the row-group level (the + * filtered reader still returns the row group of an unfiltered plan) and the final result. + */ + private void assertMatches(Table table, Predicate filter, int expectedId) throws Exception { + ReadBuilder filtered = table.newReadBuilder().withFilter(filter); + assertThat(filtered.newScan().plan().splits()).as("manifest pruning").isNotEmpty(); + + List allSplits = table.newReadBuilder().newScan().plan().splits(); + assertThat(readIds(filtered, allSplits, false)) + .as("row-group pruning") + .contains(expectedId); + + assertThat(readIds(filtered, filtered.newScan().plan().splits(), true)) + .as("filtered result") + .containsExactly(expectedId); + } + + private static List readIds( + ReadBuilder builder, List splits, boolean executeFilter) throws Exception { + List ids = new ArrayList<>(); + for (Split split : splits) { + try (RecordReader reader = + executeFilter + ? builder.newRead().executeFilter().createReader(split) + : builder.newRead().createReader(split)) { + reader.forEachRemaining(row -> ids.add(row.getInt(0))); + } + } + return ids; + } + + private Table createTable(boolean isFloat) throws Exception { + return createTable(isFloat, "t"); + } + + private Table createTable(boolean isFloat, String name) throws Exception { + DataType type = isFloat ? DataTypes.FLOAT() : DataTypes.DOUBLE(); + Options options = new Options(); + options.set("warehouse", tempDir.toUri().toString()); + try (Catalog catalog = CatalogFactory.createCatalog(CatalogContext.create(options))) { + catalog.createDatabase("db", true); + Identifier identifier = Identifier.create("db", name); + catalog.createTable( + identifier, + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("v", type) + .option("file.format", "mosaic") + .option(MosaicFileFormat.STATS_COLUMNS.key(), "v") + .option("bucket", "-1") + .build(), + false); + return catalog.getTable(identifier); + } + } + + /** Writes the values into a single file, with ids starting at 1 in write order. */ + private static void write(Table table, Object... values) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + for (int i = 0; i < values.length; i++) { + write.write(GenericRow.of(i + 1, values[i])); + } + commit.commit(write.prepareCommit()); + } + } + + private static Object value(boolean isFloat, double value) { + return isFloat ? (Object) (float) value : (Object) value; + } + + private static boolean isNativeAvailable() { + try { + Class.forName("org.apache.paimon.mosaic.NativeLib"); + return true; + } catch (Throwable t) { + return false; + } + } +} diff --git a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractorTest.java b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractorTest.java index 8477c5b06540..090f64f6c81e 100644 --- a/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractorTest.java +++ b/paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractorTest.java @@ -31,6 +31,7 @@ import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.options.Options; import org.apache.paimon.statistics.SimpleColStatsCollector; +import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -88,6 +89,18 @@ protected String fileCompression() { return "zstd"; } + @Override + protected SimpleColStats regenerate(SimpleColStats stats, DataType type) { + switch (type.getTypeRoot()) { + case FLOAT: + case DOUBLE: + // Mosaic float bounds can hide NaN or -0.0, so only the null count is kept. + return new SimpleColStats(null, null, stats.nullCount()); + default: + return stats; + } + } + @Test void testUntrackedColumnsReturnNone() throws IOException { // stats_columns only tracks f_int, but the table has f_int + f_string