Fix correctness issues in S3 Select pushdown

The IonSqlQueryBuilder would produce select queries which were not
proper transormations of the given TupleDomain, leading to incorrect
results when S3 Select was enabled.

When reading JSON files predicates like `x IS NULL` or `x IS NOT NULL`
were evaluated as `x = ''` or `x <> ''`.

When reading TextFile data the query builder ignores the table's
`null_format` field, instead assuming that null fields are encoded as
the empty string.# Please enter the commit message for your changes. Lines starting

.github/workflows/ci.yml

@@ -332,7 +332,9 @@ jobs:
           S3_BUCKET_ENDPOINT: "s3.${{ vars.TRINO_AWS_REGION }}.amazonaws.com"
         run: |
           if [ "${AWS_ACCESS_KEY_ID}" != "" ]; then
-            $MAVEN test ${MAVEN_TEST} -pl :trino-hive -P aws-tests
+            $MAVEN test ${MAVEN_TEST} -pl :trino-hive -P aws-tests \
+              -Ds3.bucket="${S3_BUCKET}" \
+              -Ds3.bucket-endpoint="${S3_BUCKET_ENDPOINT}"
           fi
       - name: Run Hive Azure ABFS Access Key Tests
         if: matrix.config != 'config-empty' # Hive 1.x does not support Azure storage

plugin/trino-hive/pom.xml

@@ -532,6 +532,7 @@
                                 <exclude>**/TestHiveGlueMetastore.java</exclude>
                                 <exclude>**/TestHiveS3AndGlueMetastoreTest.java</exclude>
                                 <exclude>**/TestTrinoS3FileSystemAwsS3.java</exclude>
+                                <exclude>**/TestS3SelectQueries.java</exclude>
                                 <exclude>**/TestFullParquetReader.java</exclude>
                                 <exclude>**/Test*FailureRecoveryTest.java</exclude>
                             </excludes>
@@ -587,6 +588,7 @@
                                 <include>**/TestHiveGlueMetastore.java</include>
                                 <include>**/TestHiveS3AndGlueMetastoreTest.java</include>
                                 <include>**/TestTrinoS3FileSystemAwsS3.java</include>
+                                <include>**/TestS3SelectQueries.java</include>
                             </includes>
                         </configuration>
                     </plugin>

plugin/trino-hive/src/main/java/io/trino/plugin/hive/s3select/IonSqlQueryBuilder.java

@@ -32,10 +32,12 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.trino.plugin.hive.s3select.S3SelectDataType.CSV;
 import static io.trino.spi.type.BigintType.BIGINT;
 import static io.trino.spi.type.BooleanType.BOOLEAN;
 import static io.trino.spi.type.DateType.DATE;
@@ -60,11 +62,27 @@ public class IonSqlQueryBuilder
     private static final String DATA_SOURCE = "S3Object s";
     private final TypeManager typeManager;
     private final S3SelectDataType s3SelectDataType;
+    private final String nullPredicate;
+    private final String notNullPredicate;
 
-    public IonSqlQueryBuilder(TypeManager typeManager, S3SelectDataType s3SelectDataType)
+    public IonSqlQueryBuilder(TypeManager typeManager, S3SelectDataType s3SelectDataType, Optional<String> optionalNullCharacterEncoding)
     {
+        if (optionalNullCharacterEncoding.isPresent()) {
+            checkArgument(s3SelectDataType == CSV, "Null character encoding should only be provided for CSV data");
+        }
+
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.s3SelectDataType = requireNonNull(s3SelectDataType, "s3SelectDataType is null");
+
+        String nullCharacterEncoding = optionalNullCharacterEncoding.orElse("");
+        this.nullPredicate = switch (s3SelectDataType) {
+            case JSON -> "IS NULL";
+            case CSV -> "= '%s'".formatted(nullCharacterEncoding);
+        };
+        this.notNullPredicate = switch (s3SelectDataType) {
+            case JSON -> "IS NOT NULL";
+            case CSV -> "!= '%s'".formatted(nullCharacterEncoding);
+        };
     }
 
     public String buildSql(List<HiveColumnHandle> columns, TupleDomain<HiveColumnHandle> tupleDomain)
@@ -143,7 +161,7 @@ private String toPredicate(Domain domain, Type type, HiveColumnHandle column)
 
         if (domain.getValues().isNone()) {
             if (domain.isNullAllowed()) {
-                return getFullyQualifiedColumnName(column) + " = '' ";
+                return getFullyQualifiedColumnName(column) + " " + nullPredicate;
             }
             return "FALSE";
         }
@@ -152,7 +170,7 @@ private String toPredicate(Domain domain, Type type, HiveColumnHandle column)
             if (domain.isNullAllowed()) {
                 return "TRUE";
             }
-            return getFullyQualifiedColumnName(column) + " <> '' ";
+            return getFullyQualifiedColumnName(column) + " " + notNullPredicate;
         }
 
         List<String> disjuncts = new ArrayList<>();
@@ -172,26 +190,35 @@ private String toPredicate(Domain domain, Type type, HiveColumnHandle column)
             }
             // If rangeConjuncts is null, then the range was ALL, which should already have been checked for
             checkState(!rangeConjuncts.isEmpty());
-            disjuncts.add("(" + Joiner.on(" AND ").join(rangeConjuncts) + ")");
+            if (rangeConjuncts.size() == 1) {
+                disjuncts.add("%s %s AND %s".formatted(getFullyQualifiedColumnName(column), notNullPredicate, getOnlyElement(rangeConjuncts)));
+            }
+            else {
+                disjuncts.add("(%s %s AND %s)".formatted(getFullyQualifiedColumnName(column), notNullPredicate, Joiner.on(" AND ").join(rangeConjuncts)));
+            }
         }
 
         // Add back all of the possible single values either as an equality or an IN predicate
         if (singleValues.size() == 1) {
-            disjuncts.add(toPredicate("=", getOnlyElement(singleValues), type, column));
+            disjuncts.add("%s %s AND %s".formatted(getFullyQualifiedColumnName(column), notNullPredicate, toPredicate("=", getOnlyElement(singleValues), type, column)));
         }
         else if (singleValues.size() > 1) {
             List<String> values = new ArrayList<>();
             for (Object value : singleValues) {
                 checkType(type);
                 values.add(valueToQuery(type, value));
             }
-            disjuncts.add(createColumn(type, column) + " IN (" + Joiner.on(",").join(values) + ")");
+            disjuncts.add("%s %s AND %s IN (%s)".formatted(
+                    getFullyQualifiedColumnName(column),
+                    notNullPredicate,
+                    createColumn(type, column),
+                    Joiner.on(",").join(values)));
         }
 
         // Add nullability disjuncts
         checkState(!disjuncts.isEmpty());
         if (domain.isNullAllowed()) {
-            disjuncts.add(getFullyQualifiedColumnName(column) + " = '' ");
+            disjuncts.add(getFullyQualifiedColumnName(column) + " " + nullPredicate);
         }
 
         return "(" + Joiner.on(" OR ").join(disjuncts) + ")";
@@ -227,7 +254,8 @@ private static String valueToQuery(Type type, Object value)
             return String.valueOf((boolean) value);
         }
         if (type.equals(DATE)) {
-            return "`" + FORMATTER.print(DAYS.toMillis((long) value)) + "`";
+            // CAST('2007-04-05T14:30Z' AS TIMESTAMP)
+            return "'" + FORMATTER.print(DAYS.toMillis((long) value)) + "'";
         }
         if (type.equals(VarcharType.VARCHAR)) {
             return "'" + ((Slice) value).toStringUtf8() + "'";
@@ -246,22 +274,11 @@ private String createColumn(Type type, HiveColumnHandle columnHandle)
         String column = getFullyQualifiedColumnName(columnHandle);
 
         if (type.equals(BIGINT) || type.equals(INTEGER) || type.equals(SMALLINT) || type.equals(TINYINT)) {
-            return formatPredicate(column, "INT");
+            return "CAST(" + column + " AS INT)";
         }
         if (type.equals(BOOLEAN)) {
-            return formatPredicate(column, "BOOL");
-        }
-        if (type.equals(DATE)) {
-            return formatPredicate(column, "TIMESTAMP");
-        }
-        if (type instanceof DecimalType decimalType) {
-            return formatPredicate(column, format("DECIMAL(%s,%s)", decimalType.getPrecision(), decimalType.getScale()));
+            return "CAST(" + column + " AS BOOL)";
         }
         return column;
     }
-
-    private String formatPredicate(String column, String type)
-    {
-        return format("case %s when '' then null else CAST(%s AS %s) end", column, column, type);
-    }
 }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/s3select/S3SelectRecordCursorProvider.java

@@ -21,6 +21,7 @@
 import io.trino.plugin.hive.HiveColumnHandle;
 import io.trino.plugin.hive.HiveRecordCursorProvider;
 import io.trino.plugin.hive.ReaderColumns;
+import io.trino.plugin.hive.s3select.csv.S3SelectCsvRecordReader;
 import io.trino.plugin.hive.type.TypeInfo;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ConnectorSession;
@@ -106,7 +107,11 @@ public Optional<ReaderRecordCursorWithProjections> createRecordCursor(
         if (s3SelectDataTypeOptional.isPresent()) {
             S3SelectDataType s3SelectDataType = s3SelectDataTypeOptional.get();
 
-            IonSqlQueryBuilder queryBuilder = new IonSqlQueryBuilder(typeManager, s3SelectDataType);
+            Optional<String> nullCharacterEncoding = Optional.empty();
+            if (s3SelectDataType == S3SelectDataType.CSV) {
+                nullCharacterEncoding = S3SelectCsvRecordReader.nullCharacterEncoding(schema);
+            }
+            IonSqlQueryBuilder queryBuilder = new IonSqlQueryBuilder(typeManager, s3SelectDataType, nullCharacterEncoding);
             String ionSqlQuery = queryBuilder.buildSql(readerColumns, effectivePredicate);
             Optional<S3SelectLineRecordReader> recordReader = S3SelectLineRecordReaderProvider.get(configuration, path, start, length, schema,
                     ionSqlQuery, s3ClientFactory, s3SelectDataType);

plugin/trino-hive/src/main/java/io/trino/plugin/hive/s3select/csv/S3SelectCsvRecordReader.java

@@ -20,9 +20,11 @@
 import com.amazonaws.services.s3.model.OutputSerialization;
 import io.trino.plugin.hive.s3select.S3SelectLineRecordReader;
 import io.trino.plugin.hive.s3select.TrinoS3ClientFactory;
+import io.trino.plugin.hive.util.SerdeConstants;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
+import java.util.Optional;
 import java.util.Properties;
 
 import static io.trino.plugin.hive.util.SerdeConstants.ESCAPE_CHAR;
@@ -104,4 +106,9 @@ public boolean shouldEnableScanRange()
                 buildInputSerialization().getCsv().getAllowQuotedRecordDelimiter());
         return CompressionType.NONE.equals(getCompressionType()) && !isQuotedRecordDelimiterAllowed;
     }
+
+    public static Optional<String> nullCharacterEncoding(Properties schema)
+    {
+        return Optional.ofNullable(schema.getProperty(SerdeConstants.SERIALIZATION_NULL_FORMAT));
+    }
 }