Reintroduce fix for dynamic pruning for with null keys in hive partition

presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java

@@ -555,7 +555,7 @@ private static boolean shouldSkipPartition(TypeManager typeManager, HiveTableLay
             HivePartitionKey hivePartitionKey = partitionKeys.get(i);
             HiveColumnHandle hiveColumnHandle = partitionColumns.get(i);
             Domain allowedDomain = domains.get(hiveColumnHandle);
-            NullableValue value = parsePartitionValue(hivePartitionKey.getName(), hivePartitionKey.getValue(), type, hiveStorageTimeZone);
+            NullableValue value = parsePartitionValue(hivePartitionKey, type, hiveStorageTimeZone);
             if (allowedDomain != null && !allowedDomain.includesNullableValue(value.getValue())) {
                 return true;
             }
@@ -605,10 +605,10 @@ public static ColumnMapping aggregated(HiveColumnHandle hiveColumnHandle, int in
             return new ColumnMapping(ColumnMappingKind.REGULAR, hiveColumnHandle, Optional.empty(), OptionalInt.of(index), Optional.empty());
         }
 
-        public static ColumnMapping prefilled(HiveColumnHandle hiveColumnHandle, String prefilledValue, Optional<HiveType> coerceFrom)
+        public static ColumnMapping prefilled(HiveColumnHandle hiveColumnHandle, Optional<String> prefilledValue, Optional<HiveType> coerceFrom)
         {
             checkArgument(hiveColumnHandle.getColumnType() == PARTITION_KEY || hiveColumnHandle.getColumnType() == SYNTHESIZED);
-            return new ColumnMapping(ColumnMappingKind.PREFILLED, hiveColumnHandle, Optional.of(prefilledValue), OptionalInt.empty(), coerceFrom);
+            return new ColumnMapping(ColumnMappingKind.PREFILLED, hiveColumnHandle, prefilledValue, OptionalInt.empty(), coerceFrom);
         }
 
         public static ColumnMapping interim(HiveColumnHandle hiveColumnHandle, int index)
@@ -634,7 +634,7 @@ public ColumnMappingKind getKind()
         public String getPrefilledValue()
         {
             checkState(kind == ColumnMappingKind.PREFILLED);
-            return prefilledValue.get();
+            return prefilledValue.orElse("\\N");
         }
 
         public HiveColumnHandle getHiveColumnHandle()

presto-hive/src/main/java/com/facebook/presto/hive/HivePartitionKey.java

@@ -17,9 +17,11 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.openjdk.jol.info.ClassLayout;
 
+import javax.annotation.Nullable;
+
 import java.util.Objects;
+import java.util.Optional;
 
-import static com.facebook.presto.hive.metastore.MetastoreUtil.HIVE_DEFAULT_DYNAMIC_PARTITION;
 import static com.google.common.base.MoreObjects.toStringHelper;
 import static java.util.Objects.requireNonNull;
 
@@ -29,18 +31,16 @@ public final class HivePartitionKey
             ClassLayout.parseClass(String.class).instanceSize() * 2;
 
     private final String name;
+    @Nullable
     private final String value;
 
     @JsonCreator
     public HivePartitionKey(
             @JsonProperty("name") String name,
-            @JsonProperty("value") String value)
+            @JsonProperty("value") Optional<String> value)
     {
-        requireNonNull(name, "name is null");
-        requireNonNull(value, "value is null");
-
-        this.name = name;
-        this.value = value.equals(HIVE_DEFAULT_DYNAMIC_PARTITION) ? "\\N" : value;
+        this.name = requireNonNull(name, "name is null");
+        this.value = requireNonNull(value, "value is null").orElse(null);
     }
 
     @JsonProperty
@@ -50,14 +50,14 @@ public String getName()
     }
 
     @JsonProperty
-    public String getValue()
+    public Optional<String> getValue()
     {
-        return value;
+        return Optional.ofNullable(value);
     }
 
     public int getEstimatedSizeInBytes()
     {
-        return INSTANCE_SIZE + name.length() * Character.BYTES + value.length() * Character.BYTES;
+        return INSTANCE_SIZE + name.length() * Character.BYTES + (value == null ? 0 : value.length() * Character.BYTES);
     }
 
     @Override

presto-hive/src/main/java/com/facebook/presto/hive/HiveUtil.java

@@ -534,10 +534,14 @@ private static boolean isValidPartitionType(Type type)
                 isCharType(type);
     }
 
+    public static NullableValue parsePartitionValue(HivePartitionKey key, Type type, DateTimeZone timeZone)
+    {
+        return parsePartitionValue(key.getName(), key.getValue().orElse(HIVE_DEFAULT_DYNAMIC_PARTITION), type, timeZone);
+    }
+
     public static NullableValue parsePartitionValue(String partitionName, String value, Type type, DateTimeZone timeZone)
     {
         verifyPartitionTypeSupported(partitionName, type);
-
         boolean isNull = HIVE_DEFAULT_DYNAMIC_PARTITION.equals(value);
 
         if (type instanceof DecimalType) {
@@ -927,19 +931,19 @@ public static String columnExtraInfo(boolean partitionKey)
         return partitionKey ? "partition key" : null;
     }
 
-    public static String getPrefilledColumnValue(HiveColumnHandle columnHandle, HivePartitionKey partitionKey, Path path, OptionalInt bucketNumber)
+    public static Optional<String> getPrefilledColumnValue(HiveColumnHandle columnHandle, HivePartitionKey partitionKey, Path path, OptionalInt bucketNumber)
     {
         if (partitionKey != null) {
             return partitionKey.getValue();
         }
         if (isPathColumnHandle(columnHandle)) {
-            return path.toString();
+            return Optional.of(path.toString());
         }
         if (isBucketColumnHandle(columnHandle)) {
             if (!bucketNumber.isPresent()) {
                 throw new PrestoException(HIVE_TABLE_BUCKETING_IS_IGNORED, "Table bucketing is ignored. The virtual \"$bucket\" column cannot be referenced.");
             }
-            return String.valueOf(bucketNumber.getAsInt());
+            return Optional.of(String.valueOf(bucketNumber.getAsInt()));
         }
         throw new PrestoException(NOT_SUPPORTED, "unsupported hidden column: " + columnHandle);
     }

presto-hive/src/main/java/com/facebook/presto/hive/PartitionLoader.java

@@ -28,6 +28,7 @@
 import static com.facebook.presto.hive.HiveErrorCode.HIVE_INVALID_METADATA;
 import static com.facebook.presto.hive.HiveErrorCode.HIVE_INVALID_PARTITION_VALUE;
 import static com.facebook.presto.hive.HiveUtil.getPartitionKeyColumnHandles;
+import static com.facebook.presto.hive.metastore.MetastoreUtil.HIVE_DEFAULT_DYNAMIC_PARTITION;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.checkCondition;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.extractPartitionValues;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -60,7 +61,7 @@ public List<HivePartitionKey> getPartitionKeys(Table table, Optional<Partition>
             }
             String value = values.get(i);
             checkCondition(value != null, HIVE_INVALID_PARTITION_VALUE, "partition key value cannot be null for field: %s", name);
-            partitionKeys.add(new HivePartitionKey(name, value));
+            partitionKeys.add(new HivePartitionKey(name, HIVE_DEFAULT_DYNAMIC_PARTITION.equals(value) ? Optional.empty() : Optional.of(value)));
         }
         return partitionKeys.build();
     }

presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java

@@ -2297,10 +2297,10 @@ public void testGetRecords()
                 HiveSplit hiveSplit = (HiveSplit) split;
 
                 List<HivePartitionKey> partitionKeys = hiveSplit.getPartitionKeys();
-                String ds = partitionKeys.get(0).getValue();
-                String fileFormat = partitionKeys.get(1).getValue();
+                String ds = partitionKeys.get(0).getValue().orElse(null);
+                String fileFormat = partitionKeys.get(1).getValue().orElse(null);
                 HiveStorageFormat fileType = HiveStorageFormat.valueOf(fileFormat.toUpperCase());
-                int dummyPartition = Integer.parseInt(partitionKeys.get(2).getValue());
+                int dummyPartition = Integer.parseInt(partitionKeys.get(2).getValue().orElse(null));
 
                 long rowNumber = 0;
                 long completedBytes = 0;
@@ -2389,10 +2389,10 @@ public void testGetPartialRecords()
                 HiveSplit hiveSplit = (HiveSplit) split;
 
                 List<HivePartitionKey> partitionKeys = hiveSplit.getPartitionKeys();
-                String ds = partitionKeys.get(0).getValue();
-                String fileFormat = partitionKeys.get(1).getValue();
+                String ds = partitionKeys.get(0).getValue().orElse(null);
+                String fileFormat = partitionKeys.get(1).getValue().orElse(null);
                 HiveStorageFormat fileType = HiveStorageFormat.valueOf(fileFormat.toUpperCase());
-                int dummyPartition = Integer.parseInt(partitionKeys.get(2).getValue());
+                int dummyPartition = Integer.parseInt(partitionKeys.get(2).getValue().orElse(null));
 
                 long rowNumber = 0;
                 try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(transaction.getTransactionHandle(), session, hiveSplit, layoutHandle, columnHandles, NON_CACHEABLE)) {

presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveFileFormats.java

@@ -118,6 +118,7 @@
 import static com.facebook.presto.tests.StructuralTestUtil.mapBlockOf;
 import static com.facebook.presto.tests.StructuralTestUtil.rowBlockOf;
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Predicates.not;
 import static com.google.common.base.Strings.padEnd;
 import static com.google.common.collect.Iterables.filter;
@@ -834,6 +835,9 @@ public TestColumn(String name, ObjectInspector objectInspector, Object writeValu
             this.writeValue = writeValue;
             this.expectedValue = expectedValue;
             this.partitionKey = partitionKey;
+            if (partitionKey) {
+                checkArgument(writeValue == null || writeValue instanceof String, "writeValue must either be null or a String value for partition keys");
+            }
         }
 
         public String getName()
@@ -856,6 +860,12 @@ public Object getWriteValue()
             return writeValue;
         }
 
+        public HivePartitionKey toHivePartitionKey()
+        {
+            checkState(partitionKey, "%s is not a partition key", this);
+            return new HivePartitionKey(name, HIVE_DEFAULT_DYNAMIC_PARTITION.equals(writeValue) ? Optional.empty() : Optional.ofNullable((String) writeValue));
+        }
+
         public Object getExpectedValue()
         {
             return expectedValue;

presto-hive/src/test/java/com/facebook/presto/hive/TestDynamicPruning.java

@@ -120,7 +120,7 @@ public void testDynamicPartitionPruning()
 
     private static ConnectorPageSource createTestingPageSource(HiveTransactionHandle transaction, HiveClientConfig config, SplitContext splitContext, MetastoreClientConfig metastoreClientConfig, File outputFile)
     {
-        ImmutableList<HivePartitionKey> partitionKeys = ImmutableList.of(new HivePartitionKey(PARTITION_COLUMN.getName(), "2020-09-09"));
+        ImmutableList<HivePartitionKey> partitionKeys = ImmutableList.of(new HivePartitionKey(PARTITION_COLUMN.getName(), Optional.of("2020-09-09")));
         Map<Integer, Column> partitionSchemaDifference = ImmutableMap.of(1, new Column("ds", HIVE_STRING, Optional.empty()));
         HiveSplit split = new HiveSplit(
                 SCHEMA_NAME,

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveDistributedJoinQueriesWithDynamicFiltering.java

@@ -112,6 +112,23 @@ public void testJoinDynamicFilteringMultiJoin()
         assertQuery(session, query, "SELECT 1, 1, 1");
     }
 
+    @Test
+    public void testJoinOnNullPartitioning()
+    {
+        assertUpdate("CREATE TABLE t3(c2 bigint, c1 bigint)");
+        assertUpdate("INSERT INTO t3 VALUES(null, 2)", 1);
+        assertUpdate("CREATE TABLE t4(c2 bigint, c1 bigint) with(partitioned_by=array['c1'])");
+        assertUpdate("INSERT INTO t4 VALUES(null, null), (2,2)", 2);
+
+        String query = "select * from t3, t4 where t3.c1=t4.c2";
+        Session session = Session.builder(getSession())
+                .setSystemProperty(ENABLE_DYNAMIC_FILTERING, "true")
+                .setSystemProperty(JOIN_DISTRIBUTION_TYPE, FeaturesConfig.JoinDistributionType.AUTOMATIC.name())
+                .setSystemProperty(JOIN_REORDERING_STRATEGY, FeaturesConfig.JoinReorderingStrategy.AUTOMATIC.name())
+                .build();
+        assertQuery(session, query, "SELECT null, 2, 2, 2");
+    }
+
     private OperatorStats searchScanFilterAndProjectOperatorStats(QueryId queryId, String tableName)
     {
         DistributedQueryRunner runner = (DistributedQueryRunner) getQueryRunner();

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveFileFormats.java

@@ -892,7 +892,7 @@ private void testCursorProvider(HiveRecordCursorProvider cursorProvider,
     {
         List<HivePartitionKey> partitionKeys = testColumns.stream()
                 .filter(TestColumn::isPartitionKey)
-                .map(input -> new HivePartitionKey(input.getName(), (String) input.getWriteValue()))
+                .map(TestColumn::toHivePartitionKey)
                 .collect(toList());
 
         List<HiveColumnHandle> partitionKeyColumnHandles = getColumnHandles(testColumns.stream().filter(TestColumn::isPartitionKey).collect(toImmutableList()));
@@ -956,7 +956,7 @@ private void testPageSourceFactory(HiveBatchPageSourceFactory sourceFactory,
     {
         List<HivePartitionKey> partitionKeys = testColumns.stream()
                 .filter(TestColumn::isPartitionKey)
-                .map(input -> new HivePartitionKey(input.getName(), (String) input.getWriteValue()))
+                .map(TestColumn::toHivePartitionKey)
                 .collect(toList());
 
         List<HiveColumnHandle> partitionKeyColumnHandles = getColumnHandles(testColumns.stream().filter(TestColumn::isPartitionKey).collect(toImmutableList()));

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplit.java

@@ -66,7 +66,7 @@ public class TestHiveSplit
     public void testJsonRoundTrip()
             throws Exception
     {
-        ImmutableList<HivePartitionKey> partitionKeys = ImmutableList.of(new HivePartitionKey("a", "apple"), new HivePartitionKey("b", "42"));
+        ImmutableList<HivePartitionKey> partitionKeys = ImmutableList.of(new HivePartitionKey("a", Optional.of("apple")), new HivePartitionKey("b", Optional.of("42")));
         ImmutableList<HostAddress> addresses = ImmutableList.of(HostAddress.fromParts("127.0.0.1", 44), HostAddress.fromParts("127.0.0.1", 45));
         Map<String, String> customSplitInfo = ImmutableMap.of("key", "value");
         Set<ColumnHandle> redundantColumnDomains = ImmutableSet.of(new HiveColumnHandle(

presto-hive/src/test/java/com/facebook/presto/hive/TestOrcBatchPageSourceMemoryTracking.java

@@ -400,7 +400,7 @@ public TestPreparer(String tempFilePath, List<TestColumn> testColumns, int numRo
 
             partitionKeys = testColumns.stream()
                     .filter(TestColumn::isPartitionKey)
-                    .map(input -> new HivePartitionKey(input.getName(), (String) input.getWriteValue()))
+                    .map(input -> new HivePartitionKey(input.getName(), Optional.ofNullable((String) input.getWriteValue())))
                     .collect(toList());
 
             table = new TableHandle(

presto-hive/src/test/java/com/facebook/presto/hive/statistics/TestMetastoreHiveStatisticsProvider.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive.statistics;
 
+import com.facebook.presto.common.predicate.NullableValue;
 import com.facebook.presto.common.type.DecimalType;
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.hive.HiveBasicStatistics;
@@ -37,6 +38,7 @@
 import com.facebook.presto.testing.TestingConnectorSession;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import io.airlift.slice.Slices;
 import org.joda.time.DateTimeZone;
 import org.testng.annotations.Test;
 
@@ -87,6 +89,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
 
 public class TestMetastoreHiveStatisticsProvider
 {
@@ -116,6 +120,16 @@ public void testGetPartitionsSample()
         assertEquals(getPartitionsSample(ImmutableList.of(p1, p2, p3, p4, p5), 3), ImmutableList.of(p1, p5, p4));
     }
 
+    @Test
+    public void testNullablePartitionValue()
+    {
+        HivePartition partitionWithNull = partition("p1=__HIVE_DEFAULT_PARTITION__/p2=1");
+        assertTrue(partitionWithNull.getKeys().containsValue(new NullableValue(VARCHAR, null)));
+        HivePartition partitionNameWithSlashN = partition("p1=\\N/p2=2");
+        assertTrue(partitionNameWithSlashN.getKeys().containsValue(new NullableValue(VARCHAR, Slices.utf8Slice("\\N"))));
+        assertFalse(partitionNameWithSlashN.getKeys().containsValue(new NullableValue(VARCHAR, null)));
+    }
+
     @Test
     public void testValidatePartitionStatistics()
     {