diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java index b6640884ff2d..208ec6110276 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithHiveTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.hive.TestHiveMetastore; import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import org.junit.jupiter.api.AfterAll; @@ -32,6 +33,7 @@ import java.io.FileNotFoundException; import java.io.IOException; +import java.net.URL; import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; @@ -116,6 +118,9 @@ public void testCreateFormatTable() throws IOException { @Test public void testSpecifyHiveConfDirInGenericCatalog() throws IOException { + URL hiveSiteLocation = HiveConf.getHiveSiteLocation(); + boolean loadMetastoreConfig = HiveConf.isLoadMetastoreConfig(); + boolean loadHiveServer2Config = HiveConf.isLoadHiveServer2Config(); try (SparkSession spark = createSessionBuilder() .config("spark.sql.catalog.spark_catalog.hive-conf-dir", "nonExistentPath") @@ -127,6 +132,10 @@ public void testSpecifyHiveConfDirInGenericCatalog() throws IOException { .rootCause() .isInstanceOf(FileNotFoundException.class) .hasMessageContaining("nonExistentPath"); + } finally { + HiveConf.setHiveSiteLocation(hiveSiteLocation); + HiveConf.setLoadMetastoreConfig(loadMetastoreConfig); + HiveConf.setLoadHiveServer2Config(loadHiveServer2Config); } } diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java index 67282c84a9f6..90fd081ffdf6 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkGenericCatalogWithHiveTest.java @@ -21,6 +21,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.hive.TestHiveMetastore; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import org.junit.jupiter.api.AfterAll; @@ -30,6 +31,7 @@ import java.io.FileNotFoundException; import java.io.IOException; +import java.net.URL; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -185,7 +187,10 @@ public void testBuildWithHive(@TempDir java.nio.file.Path tempDir) throws IOExce @Test public void testHiveCatalogOptions(@TempDir java.nio.file.Path tempDir) throws IOException { Path warehousePath = new Path("file:" + tempDir.toString()); - SparkSession spark = + URL hiveSiteLocation = HiveConf.getHiveSiteLocation(); + boolean loadMetastoreConfig = HiveConf.isLoadMetastoreConfig(); + boolean loadHiveServer2Config = HiveConf.isLoadHiveServer2Config(); + try (SparkSession spark = SparkSession.builder() .config("spark.sql.catalog.spark_catalog.hive-conf-dir", "nonExistentPath") .config("spark.sql.warehouse.dir", warehousePath.toString()) @@ -198,13 +203,15 @@ public void testHiveCatalogOptions(@TempDir java.nio.file.Path tempDir) throws I "spark.sql.extensions", "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions") .master("local[2]") - .getOrCreate(); - - assertThatThrownBy(() -> spark.sql("CREATE DATABASE my_db")) - .rootCause() - .isInstanceOf(FileNotFoundException.class) - .hasMessageContaining("nonExistentPath"); - - spark.close(); + .getOrCreate()) { + assertThatThrownBy(() -> spark.sql("CREATE DATABASE my_db")) + .rootCause() + .isInstanceOf(FileNotFoundException.class) + .hasMessageContaining("nonExistentPath"); + } finally { + HiveConf.setHiveSiteLocation(hiveSiteLocation); + HiveConf.setLoadMetastoreConfig(loadMetastoreConfig); + HiveConf.setLoadHiveServer2Config(loadHiveServer2Config); + } } } diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java index 02326ba684af..66c12b801c8c 100644 --- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java +++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexPredicateConverter.java @@ -38,6 +38,7 @@ import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.predicate.PredicateVisitor; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.LocalZonedTimestampType; import org.apache.paimon.types.TimestampType; @@ -80,6 +81,12 @@ public Expression visit(LeafPredicate predicate) { return Expression.isNotNull(field); } + if (fieldRef.type().isAnyOf(DataTypeRoot.FLOAT, DataTypeRoot.DOUBLE)) { + // Vortex orders NaN signs and payloads differently from Paimon. Even a finite + // literal can therefore exclude matching NaN rows. Null checks remain safe. + return null; + } + List literals = predicate.literals(); if (literals == null || literals.isEmpty()) { return null; @@ -161,10 +168,6 @@ private static Expression toLiteral(DataType type, Object value) { return Expression.literalDate((Integer) value, Expression.TimeUnit.DAYS); case BIGINT: return Expression.literal((Long) value); - case FLOAT: - return Expression.literal((Float) value); - case DOUBLE: - return Expression.literal((Double) value); case CHAR: case VARCHAR: return Expression.literal( diff --git a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java index 222e6bd0ce62..52da70021a1c 100644 --- a/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java +++ b/paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexPredicateConverterTest.java @@ -43,12 +43,16 @@ import dev.vortex.api.Expression; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.math.BigDecimal; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Set; +import java.util.TreeSet; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -515,6 +519,205 @@ public void testTimestampNanosOverflowingLiteralNotPushed(@TempDir java.nio.file assertEquals(0L, rows.get(0).getTimestamp(0, 0).getMillisecond()); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testFloatingPointValuePredicatesNotPushed(boolean isFloat) { + PredicateBuilder builder = new PredicateBuilder(floatingPointRowType(isFloat)); + for (Predicate predicate : floatingPointPredicates(builder, isFloat)) { + try (Expression expression = + VortexPredicateConverter.toVortexExpression( + Collections.singletonList(predicate))) { + assertNull(expression, predicate.toString()); + } + } + + // Null checks do not compare floating-point values and remain safe to push down. + for (Predicate predicate : Arrays.asList(builder.isNull(1), builder.isNotNull(1))) { + try (Expression expression = + VortexPredicateConverter.toVortexExpression( + Collections.singletonList(predicate))) { + assertValidExpression(expression); + } + } + + Predicate integer = builder.greaterThan(0, 3); + Predicate floating = builder.greaterThan(1, floatingPointNumber(isFloat, 0.0)); + for (List children : + Arrays.asList(Arrays.asList(integer, floating), Arrays.asList(floating, integer))) { + try (Expression and = + VortexPredicateConverter.toVortexExpression( + Collections.singletonList(PredicateBuilder.and(children))); + Expression or = + VortexPredicateConverter.toVortexExpression( + Collections.singletonList(PredicateBuilder.or(children)))) { + assertValidExpression(and); + assertNull(or); + } + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testFloatingPointPredicatesPreserveMatchingRows( + boolean isFloat, @TempDir java.nio.file.Path tempDir) throws Exception { + RowType rowType = floatingPointRowType(isFloat); + PredicateBuilder builder = new PredicateBuilder(rowType); + List values = + Arrays.asList( + floatingPointNumber(isFloat, Double.NEGATIVE_INFINITY), + floatingPointNumber(isFloat, -12.5), + floatingPointNumber(isFloat, -1.0), + floatingPointNumber(isFloat, -0.0), + floatingPointNumber(isFloat, 0.0), + floatingPointNumber(isFloat, 1.0), + floatingPointNumber(isFloat, 12.5), + floatingPointNumber(isFloat, Double.POSITIVE_INFINITY), + floatingPointNaN(isFloat, false, 0), + floatingPointNaN(isFloat, false, 1), + floatingPointNaN(isFloat, false, 0x1234), + floatingPointNaN(isFloat, true, 1), + floatingPointNaN(isFloat, true, 0x1234), + null); + VortexFileFormat format = + new VortexFileFormatFactory() + .create(new FileFormatFactory.FormatContext(new Options(), 1024, 1024)); + FileIO fileIO = new LocalFileIO(); + Path file = new Path(tempDir.resolve("floating_point_predicates.vortex").toUri()); + try (FormatWriter writer = + ((SupportsDirectWrite) format.createWriterFactory(rowType)) + .create(fileIO, file, "")) { + for (int i = 0; i < values.size(); i++) { + writer.addElement(GenericRow.of(i, values.get(i))); + } + } + + List allRows = + readFloatingPointRows(format, rowType, fileIO, file, null, isFloat); + assertEquals(values.size(), allRows.size()); + // Paimon considers all NaN signs and payloads equal, unlike Vortex comparisons. + assertEquals( + new TreeSet<>(Arrays.asList(8, 9, 10, 11, 12)), + matchingIds(allRows, builder.equal(1, floatingPointNaN(isFloat, true, 1)))); + + List predicates = floatingPointPredicates(builder, isFloat); + predicates.add(builder.isNull(1)); + predicates.add(builder.isNotNull(1)); + Predicate integer = builder.lessThan(0, 12); + Predicate floating = builder.greaterThan(1, floatingPointNumber(isFloat, 0.0)); + predicates.add(integer); + for (List children : + Arrays.asList(Arrays.asList(integer, floating), Arrays.asList(floating, integer))) { + predicates.add(PredicateBuilder.and(children)); + predicates.add(PredicateBuilder.or(children)); + } + for (Predicate predicate : predicates) { + List candidates = + readFloatingPointRows(format, rowType, fileIO, file, predicate, isFloat); + Set expected = matchingIds(allRows, predicate); + // The direct format reader supplies candidates; the caller applies the residual. + assertTrue( + matchingIds(candidates, PredicateBuilder.alwaysTrue()).containsAll(expected), + predicate.toString()); + assertEquals(expected, matchingIds(candidates, predicate), predicate.toString()); + } + } + + private static RowType floatingPointRowType(boolean isFloat) { + return RowType.builder() + .field("id", DataTypes.INT()) + .field("value", isFloat ? DataTypes.FLOAT() : DataTypes.DOUBLE()) + .build(); + } + + private static List floatingPointPredicates( + PredicateBuilder builder, boolean isFloat) { + List predicates = new ArrayList<>(); + // A finite literal is also unsafe: the column can contain a negative NaN. + for (Object literal : + Arrays.asList( + floatingPointNumber(isFloat, 0.0), + floatingPointNaN(isFloat, false, 0x1234), + floatingPointNaN(isFloat, true, 1), + null)) { + predicates.add(builder.equal(1, literal)); + predicates.add(builder.notEqual(1, literal)); + predicates.add(builder.greaterThan(1, literal)); + predicates.add(builder.greaterOrEqual(1, literal)); + predicates.add(builder.lessThan(1, literal)); + predicates.add(builder.lessOrEqual(1, literal)); + } + // Small memberships expand to comparisons; larger ones remain membership leaves. + for (int size : new int[] {1, 20, 21}) { + List literals = new ArrayList<>(); + literals.add(floatingPointNaN(isFloat, true, 1)); + for (int i = 1; i < size; i++) { + literals.add(floatingPointNumber(isFloat, i)); + } + predicates.add(builder.in(1, literals)); + predicates.add(builder.notIn(1, literals)); + } + return predicates; + } + + private static Object floatingPointNumber(boolean isFloat, double value) { + if (isFloat) { + return (float) value; + } + return value; + } + + private static Object floatingPointNaN(boolean isFloat, boolean negative, int payload) { + if (isFloat) { + return Float.intBitsToFloat((negative ? 0xffc00000 : 0x7fc00000) | payload); + } + return Double.longBitsToDouble( + (negative ? 0xfff8000000000000L : 0x7ff8000000000000L) | payload); + } + + private static Set matchingIds(List rows, Predicate predicate) { + Set ids = new TreeSet<>(); + for (InternalRow row : rows) { + if (predicate.test(row)) { + ids.add(row.getInt(0)); + } + } + return ids; + } + + private static List readFloatingPointRows( + VortexFileFormat format, + RowType rowType, + FileIO fileIO, + Path file, + Predicate predicate, + boolean isFloat) + throws Exception { + List predicates = + predicate == null ? Collections.emptyList() : Collections.singletonList(predicate); + RecordReader reader = + format.createReaderFactory(rowType, rowType, predicates) + .createReader( + new FormatReaderContext( + fileIO, file, fileIO.getFileSize(file), null, null)); + List rows = new ArrayList<>(); + try (RecordReaderIterator iterator = new RecordReaderIterator<>(reader)) { + while (iterator.hasNext()) { + InternalRow row = iterator.next(); + Object value = null; + if (!row.isNullAt(1)) { + // Copy directly so a binary serializer cannot normalize NaN payloads. + if (isFloat) { + value = row.getFloat(1); + } else { + value = row.getDouble(1); + } + } + rows.add(GenericRow.of(row.getInt(0), value)); + } + } + return rows; + } + private List roundTrip( java.nio.file.Path tempDir, RowType rowType,