Implement visitMetadataDelete in Planner Optimizations (#25745)

Summary:
Pull Request resolved: #25745

Early Metadata delete optimization from Metalake isn't working without proper support in Planner Optimization

Failure E.g
https://www.internalfb.com/intern/presto/query/?query_id=20250807_004822_04759_jyfv8#stack_trace

- 2 partition + rand  https://www.internalfb.com/intern/presto/query/?query_id=20250808_205037_00219_ykr22#sql
- 1 partition + rand  https://www.internalfb.com/intern/presto/query/?query_id=20250808_205410_00240_ykr22#sql
- 1 partition + rand +  between - https://www.internalfb.com/intern/presto/query/?query_id=20250808_205243_00239_ykr22#sql
- 1 partition + between https://www.internalfb.com/intern/presto/query/?query_id=20250808_205610_00259_ykr22#sql

Add handling logic for visit Metadata Delete
Reference - https://www.internalfb.com/diff/D66677607

Reviewed By: ghelmling

Differential Revision: D80040451

Author: natashasehgal

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/AddExchanges.java

@@ -38,6 +38,7 @@
 import com.facebook.presto.spi.plan.JoinNode;
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
+import com.facebook.presto.spi.plan.MetadataDeleteNode;
 import com.facebook.presto.spi.plan.OutputNode;
 import com.facebook.presto.spi.plan.Partitioning;
 import com.facebook.presto.spi.plan.PartitioningHandle;
@@ -732,6 +733,13 @@ public PlanWithProperties visitTableScan(TableScanNode node, PreferredProperties
             return planTableScan(node, TRUE_CONSTANT);
         }
 
+        @Override
+        public PlanWithProperties visitMetadataDelete(MetadataDeleteNode node, PreferredProperties preferredProperties)
+        {
+            // MetadataDeleteNode is a leaf node that runs on coordinator
+            return new PlanWithProperties(node);
+        }
+
         @Override
         public PlanWithProperties visitTableWriter(TableWriterNode node, PreferredProperties preferredProperties)
         {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/PropertyDerivations.java

@@ -34,6 +34,7 @@
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
 import com.facebook.presto.spi.plan.MergeJoinNode;
+import com.facebook.presto.spi.plan.MetadataDeleteNode;
 import com.facebook.presto.spi.plan.OrderingScheme;
 import com.facebook.presto.spi.plan.OutputNode;
 import com.facebook.presto.spi.plan.PartitioningHandle;
@@ -391,6 +392,15 @@ public ActualProperties visitUpdate(UpdateNode node, List<ActualProperties> inpu
             return Iterables.getOnlyElement(inputProperties).translateVariable(symbol -> Optional.empty());
         }
 
+        @Override
+        public ActualProperties visitMetadataDelete(MetadataDeleteNode node, List<ActualProperties> inputProperties)
+        {
+            // MetadataDeleteNode runs on coordinator and produces row count
+            return ActualProperties.builder()
+                    .global(coordinatorSingleStreamPartition())
+                    .build();
+        }
+
         @Override
         public ActualProperties visitJoin(JoinNode node, List<ActualProperties> inputProperties)
         {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/StreamPropertyDerivations.java

@@ -27,6 +27,7 @@
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
 import com.facebook.presto.spi.plan.MergeJoinNode;
+import com.facebook.presto.spi.plan.MetadataDeleteNode;
 import com.facebook.presto.spi.plan.OutputNode;
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.spi.plan.ProjectNode;
@@ -462,6 +463,13 @@ public StreamProperties visitUpdate(UpdateNode node, List<StreamProperties> inpu
             return properties.withUnspecifiedPartitioning();
         }
 
+        @Override
+        public StreamProperties visitMetadataDelete(MetadataDeleteNode node, List<StreamProperties> inputProperties)
+        {
+            // MetadataDeleteNode runs on coordinator and outputs a single row count
+            return StreamProperties.singleStream();
+        }
+
         @Override
         public StreamProperties visitTableWriteMerge(TableWriterMergeNode node, List<StreamProperties> inputProperties)
         {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/UnaliasSymbolReferences.java

@@ -34,6 +34,7 @@
 import com.facebook.presto.spi.plan.JoinNode;
 import com.facebook.presto.spi.plan.LimitNode;
 import com.facebook.presto.spi.plan.MarkDistinctNode;
+import com.facebook.presto.spi.plan.MetadataDeleteNode;
 import com.facebook.presto.spi.plan.Ordering;
 import com.facebook.presto.spi.plan.OrderingScheme;
 import com.facebook.presto.spi.plan.OutputNode;
@@ -707,6 +708,13 @@ public PlanNode visitTableWriteMerge(TableWriterMergeNode node, RewriteContext<V
             return mapper.map(node, source);
         }
 
+        @Override
+        public PlanNode visitMetadataDelete(MetadataDeleteNode node, RewriteContext<Void> context)
+        {
+            // MetadataDeleteNode has no symbols to unalias, so return unchanged
+            return node;
+        }
+
         @Override
         public PlanNode visitPlan(PlanNode node, RewriteContext<Void> context)
         {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/plan/InternalPlanVisitor.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.sql.planner.plan;
 
+import com.facebook.presto.spi.plan.MetadataDeleteNode;
 import com.facebook.presto.spi.plan.PlanVisitor;
 import com.facebook.presto.sql.planner.CanonicalJoinNode;
 import com.facebook.presto.sql.planner.CanonicalTableScanNode;
@@ -126,4 +127,9 @@ public R visitSequence(SequenceNode node, C context)
     {
         return visitPlan(node, context);
     }
+
+    public R visitMetadataDelete(MetadataDeleteNode node, C context)
+    {
+        return visitPlan(node, context);
+    }
 }