Remove redundant metastore call from IcebergPageSourceProvider

godaridillibabu · bibith4 · commit 98246e21db49 · 2025-11-29T15:46:32.000+05:30
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java
@@ -372,6 +372,7 @@ public ConnectorTableLayoutResult getTableLayoutForConstraint(
                         .setPartitionColumnPredicate(partitionColumnPredicate.simplify())
                         .setPartitions(Optional.ofNullable(partitions.size() == 0 ? null : partitions))
                         .setTable(handle)
+                        .setTableLocation(Optional.of(icebergTable.location()))
                         .build());
         return new ConnectorTableLayoutResult(layout, constraint.getSummary());
     }
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSourceProvider.java
@@ -79,7 +79,6 @@
 import com.facebook.presto.spi.FixedPageSource;
 import com.facebook.presto.spi.PageIndexerFactory;
 import com.facebook.presto.spi.PrestoException;
-import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.SplitContext;
 import com.facebook.presto.spi.connector.ConnectorPageSourceProvider;
 import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
@@ -753,21 +752,18 @@ public ConnectorPageSource createPageSource(
         IcebergSplit split = (IcebergSplit) connectorSplit;
         IcebergTableHandle table = icebergLayout.getTable();
         if (similaritySearchEnabled && split.isAnn()) {
-            SchemaTableName schemaTableName = table.getSchemaTableName();
-            Table icebergTable = IcebergUtil.getHiveIcebergTable(
-                    metastore,
-                    hdfsEnvironment,
-                    tableOperationsConfig,
-                    manifestFileCache,
-                    session,
-                    catalogName,
-                    schemaTableName);
+            String tableLocation = icebergLayout.getTableLocation()
+                    .orElseThrow(() -> new IllegalStateException("Table location is required for ANN queries"));
+
+            HdfsContext hdfsContext = new HdfsContext(session, table.getSchemaName(), table.getTableName(), split.getPath(), false);
+            HdfsFileIO hdfsFileIO = new HdfsFileIO(manifestFileCache, hdfsEnvironment, hdfsContext);
 
             return new ANNPageSource(
                     new FixedPageSource(ImmutableList.of()),
                     split.getQueryVector(),
                     split.getTopN(),
-                    icebergTable);
+                    tableLocation,
+                    hdfsFileIO);
         }
 
         List<ColumnHandle> columns = desiredColumns;
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergTableLayoutHandle.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergTableLayoutHandle.java
@@ -42,6 +42,7 @@ public class IcebergTableLayoutHandle
     private final Map<String, IcebergColumnHandle> predicateColumns;
     private final Optional<Set<IcebergColumnHandle>> requestedColumns;
     private final IcebergTableHandle table;
+    private final Optional<String> tableLocation;
 
     @JsonCreator
     public IcebergTableLayoutHandle(
@@ -53,7 +54,8 @@ public IcebergTableLayoutHandle(
             @JsonProperty("requestedColumns") Optional<Set<IcebergColumnHandle>> requestedColumns,
             @JsonProperty("pushdownFilterEnabled") boolean pushdownFilterEnabled,
             @JsonProperty("partitionColumnPredicate") TupleDomain<ColumnHandle> partitionColumnPredicate,
-            @JsonProperty("table") IcebergTableHandle table)
+            @JsonProperty("table") IcebergTableHandle table,
+            @JsonProperty("tableLocation") Optional<String> tableLocation)
     {
         this(
                 partitionColumns.stream().map(BaseHiveColumnHandle.class::cast).collect(toList()),
@@ -65,7 +67,8 @@ public IcebergTableLayoutHandle(
                 pushdownFilterEnabled,
                 partitionColumnPredicate,
                 Optional.empty(),
-                table);
+                table,
+                tableLocation);
     }
 
     public IcebergTableLayoutHandle(
@@ -78,7 +81,8 @@ public IcebergTableLayoutHandle(
             boolean pushdownFilterEnabled,
             TupleDomain<ColumnHandle> partitionColumnPredicate,
             Optional<List<HivePartition>> partitions,
-            IcebergTableHandle table)
+            IcebergTableHandle table,
+            Optional<String> tableLocation)
     {
         super(
                 partitionColumns,
@@ -92,6 +96,7 @@ public IcebergTableLayoutHandle(
         this.predicateColumns = requireNonNull(predicateColumns, "predicateColumns is null");
         this.requestedColumns = requireNonNull(requestedColumns, "requestedColumns is null");
         this.table = requireNonNull(table, "table is null");
+        this.tableLocation = requireNonNull(tableLocation, "tableLocation is null");
     }
 
     @JsonProperty
@@ -118,6 +123,12 @@ public IcebergTableHandle getTable()
         return table;
     }
 
+    @JsonProperty
+    public Optional<String> getTableLocation()
+    {
+        return tableLocation;
+    }
+
     public TupleDomain<IcebergColumnHandle> getValidPredicate()
     {
         TupleDomain<IcebergColumnHandle> predicate = getDomainPredicate()
@@ -173,6 +184,7 @@ public static class Builder
         private TupleDomain<ColumnHandle> partitionColumnPredicate;
         private Optional<List<HivePartition>> partitions;
         private IcebergTableHandle table;
+        private Optional<String> tableLocation = Optional.empty();
 
         public Builder setPartitionColumns(List<BaseHiveColumnHandle> partitionColumns)
         {
@@ -234,6 +246,12 @@ public Builder setTable(IcebergTableHandle table)
             return this;
         }
 
+        public Builder setTableLocation(Optional<String> tableLocation)
+        {
+            this.tableLocation = tableLocation;
+            return this;
+        }
+
         public IcebergTableLayoutHandle build()
         {
             return new IcebergTableLayoutHandle(
@@ -246,7 +264,8 @@ public IcebergTableLayoutHandle build()
                     pushdownFilterEnabled,
                     partitionColumnPredicate,
                     partitions,
-                    table);
+                    table,
+                    tableLocation);
         }
     }
 }
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/optimizer/IcebergFilterPushdown.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/optimizer/IcebergFilterPushdown.java
@@ -178,6 +178,7 @@ protected ConnectorPushdownFilterResult getConnectorPushdownFilterResult(
                                     .setPartitionColumnPredicate(partitionColumnPredicate)
                                     .setPartitions(Optional.ofNullable(partitions.size() == 0 ? null : partitions))
                                     .setTable((IcebergTableHandle) tableHandle)
+                                    .setTableLocation(Optional.of(icebergTable.location()))
                                     .build()),
                     remainingExpressions.getDynamicFilterExpression());
         }
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/optimizer/IcebergPlanOptimizer.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/optimizer/IcebergPlanOptimizer.java
@@ -227,7 +227,8 @@ public PlanNode visitFilter(FilterNode filter, RewriteContext<Void> context)
                             identityPartitionColumnPredicate.simplify()
                                     .intersect(icebergTableLayoutHandle.getPartitionColumnPredicate()),
                             icebergTableLayoutHandle.getPartitions(),
-                            icebergTableLayoutHandle.getTable()));
+                            icebergTableLayoutHandle.getTable(),
+                            icebergTableLayoutHandle.getTableLocation()));
             TableScanNode newTableScan = new TableScanNode(
                     tableScan.getSourceLocation(),
                     tableScan.getId(),
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/tvf/ANNPageSource.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/tvf/ANNPageSource.java
@@ -32,7 +32,6 @@
 import io.github.jbellis.jvector.vector.VectorizationProvider;
 import io.github.jbellis.jvector.vector.types.VectorFloat;
 import io.github.jbellis.jvector.vector.types.VectorTypeSupport;
-import org.apache.iceberg.Table;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -49,17 +48,19 @@ public class ANNPageSource
     private final List<Float> queryVector;
     private final int topN;
     private boolean finished;
-    private Table icebergTable;
+    private final String tableLocation;
+    private final HdfsFileIO hdfsFileIO;
     private static final VectorTypeSupport vts = VectorizationProvider.getInstance().getVectorTypeSupport();
     private static final String VECTOR_INDEX_DIR = ".vector_index";
     private static final Logger log = Logger.get(ANNPageSource.class);
 
-    public ANNPageSource(ConnectorPageSource delegate, List<Float> queryVector, int topN, Table icebergTable)
+    public ANNPageSource(ConnectorPageSource delegate, List<Float> queryVector, int topN, String tableLocation, HdfsFileIO hdfsFileIO)
     {
         this.delegate = delegate;
         this.queryVector = queryVector;
         this.topN = topN;
-        this.icebergTable = icebergTable;
+        this.tableLocation = tableLocation;
+        this.hdfsFileIO = hdfsFileIO;
     }
 
     @Override
@@ -92,7 +93,6 @@ public Page getNextPage()
         if (finished) {
             return null;
         }
-        String tableLocation = icebergTable.location();
         log.info("tableLocation: %s", tableLocation);
 
         String indexDirPath = tableLocation + "/" + VECTOR_INDEX_DIR;
@@ -102,7 +102,6 @@ public Page getNextPage()
         log.info("Loading node-to-row mapping from: %s", mappingPath);
         java.nio.file.Path tempIndexFile = null;
         try {
-            HdfsFileIO hdfsFileIO = (HdfsFileIO) icebergTable.io();
             // Load mapping directly from stream
             NodeRowIdMapping mapping;
             try {

Original file line number	Diff line number	Diff line change
`@@ -372,6 +372,7 @@ public ConnectorTableLayoutResult getTableLayoutForConstraint(`
`372`	`372`	`.setPartitionColumnPredicate(partitionColumnPredicate.simplify())`
`373`	`373`	`.setPartitions(Optional.ofNullable(partitions.size() == 0 ? null : partitions))`
`374`	`374`	`.setTable(handle)`
	`375`	`+ .setTableLocation(Optional.of(icebergTable.location()))`
`375`	`376`	`.build());`
`376`	`377`	`return new ConnectorTableLayoutResult(layout, constraint.getSummary());`
`377`	`378`	`}`
Original file line number	Diff line number	Diff line change
`@@ -178,6 +178,7 @@ protected ConnectorPushdownFilterResult getConnectorPushdownFilterResult(`
`178`	`178`	`.setPartitionColumnPredicate(partitionColumnPredicate)`
`179`	`179`	`.setPartitions(Optional.ofNullable(partitions.size() == 0 ? null : partitions))`
`180`	`180`	`.setTable((IcebergTableHandle) tableHandle)`
	`181`	`+ .setTableLocation(Optional.of(icebergTable.location()))`
`181`	`182`	`.build()),`
`182`	`183`	`remainingExpressions.getDynamicFilterExpression());`
`183`	`184`	`}`