WIP feat: Add TopN optimization for Pinot split provider

Implement TopN pushdown optimization in ClpPinotSplitProvider to minimize
archive scanning when executing LIMIT + ORDER BY queries on timestamp
metadata columns.

The optimization uses archive metadata (timestamp bounds and message counts)
to intelligently select only the minimum set of archives needed to guarantee
correct TopN results, significantly reducing I/O for time-range queries.

Implementation details:

Archive selection algorithm:
- Fetches archive metadata (file path, creation time, modification time, message count)
- Merges overlapping archives into non-overlapping groups by timestamp ranges
- For DESC ordering: includes newest group + older groups until limit is covered
- For ASC ordering: includes oldest group + newer groups until limit is covered
- Uses worst-case analysis to ensure correctness when archives overlap

Code structure:
- ClpPinotSplitProvider.listSplits(): Detects TopN specs and routes to optimized path
- selectTopNArchives(): Implements the archive selection algorithm
- toArchiveGroups(): Merges overlapping archives into logical components
- ArchiveMeta: Represents archive metadata with validation
- ArchiveGroup: Represents merged archive groups for overlap handling

ClpTopNSpec enhancements:
- Made fields private with proper encapsulation
- Added @JsonProperty annotations to getters for correct serialization
- Maintains immutability for thread-safety

Code quality improvements:
- Proper exception handling with context-aware error messages
- Input validation in constructors (bounds checking, null checks)
- Extracted determineSplitType() helper to eliminate duplication
- Made fields final where immutable (pinotDatabaseUrl, ArchiveMeta fields)
- Improved logging with table names and SQL queries for debugging
- Better encapsulation: private fields, getters-only access

Performance impact:
For queries like "SELECT * FROM logs ORDER BY timestamp DESC LIMIT 100",
this eliminates scanning of archives outside the time range of interest,
providing substantial I/O reduction for large archive sets.

Author: jackluo923

presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java

@@ -45,19 +45,21 @@ public static final class Ordering
         private final Order order;
 
         @JsonCreator
-        public Ordering(@JsonProperty("columns") List<String> columns, @JsonProperty("order") Order order)
+        public Ordering(
+                @JsonProperty("columns") List<String> columns,
+                @JsonProperty("order") Order order)
         {
             this.columns = requireNonNull(columns, "column is null");
             this.order = requireNonNull(order, "order is null");
         }
 
-        @JsonProperty
+        @JsonProperty("columns")
         public List<String> getColumns()
         {
             return columns;
         }
 
-        @JsonProperty
+        @JsonProperty("order")
         public Order getOrder()
         {
             return order;
@@ -93,7 +95,9 @@ public String toString()
     private final List<Ordering> orderings;
 
     @JsonCreator
-    public ClpTopNSpec(@JsonProperty("limit") long limit, @JsonProperty("orderings") List<Ordering> orderings)
+    public ClpTopNSpec(
+            @JsonProperty("limit") long limit,
+            @JsonProperty("orderings") List<Ordering> orderings)
     {
         if (limit <= 0) {
             throw new IllegalArgumentException("limit must be > 0");
@@ -105,13 +109,13 @@ public ClpTopNSpec(@JsonProperty("limit") long limit, @JsonProperty("orderings")
         this.orderings = orderings;
     }
 
-    @JsonProperty
+    @JsonProperty("limit")
     public long getLimit()
     {
         return limit;
     }
 
-    @JsonProperty
+    @JsonProperty("orderings")
     public List<Ordering> getOrderings()
     {
         return orderings;

presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.plugin.clp.ClpSplit;
 import com.facebook.presto.plugin.clp.ClpTableHandle;
 import com.facebook.presto.plugin.clp.ClpTableLayoutHandle;
+import com.facebook.presto.plugin.clp.optimization.ClpTopNSpec;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
@@ -28,39 +29,241 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 
 import static com.facebook.presto.plugin.clp.ClpSplit.SplitType;
 import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.ARCHIVE;
 import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.IR;
 import static java.lang.String.format;
+import static java.util.Comparator.comparingLong;
+import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 public class ClpPinotSplitProvider
         implements ClpSplitProvider
 {
     private static final Logger log = Logger.get(ClpPinotSplitProvider.class);
     private static final String SQL_SELECT_SPLITS_TEMPLATE = "SELECT tpath FROM %s WHERE 1 = 1 AND (%s) LIMIT 999999";
+    private static final String SQL_SELECT_SPLIT_META_TEMPLATE = "SELECT tpath, creationtime, lastmodifiedtime, num_messages FROM %s WHERE 1 = 1 AND (%s) ORDER BY %s %s LIMIT 999999";
     private final ClpConfig config;
+    private final URL pinotDatabaseUrl;
 
     @Inject
     public ClpPinotSplitProvider(ClpConfig config)
     {
-        this.config = config;
+        this.config = requireNonNull(config, "config is null");
+        try {
+            this.pinotDatabaseUrl = new URL(config.getMetadataDbUrl() + "/query/sql");
+        }
+        catch (MalformedURLException e) {
+            throw new IllegalArgumentException(
+                    format("Invalid Pinot database URL: %s/query/sql", config.getMetadataDbUrl()), e);
+        }
     }
 
     @Override
     public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
     {
-        ImmutableList.Builder<ClpSplit> splits = new ImmutableList.Builder<>();
         ClpTableHandle clpTableHandle = clpTableLayoutHandle.getTable();
+        Optional<ClpTopNSpec> topNSpecOptional = clpTableLayoutHandle.getTopN();
         String tableName = clpTableHandle.getSchemaTableName().getTableName();
         try {
-            URL url = new URL(config.getMetadataDbUrl() + "/query/sql");
+            ImmutableList.Builder<ClpSplit> splits = new ImmutableList.Builder<>();
+            if (topNSpecOptional.isPresent()) {
+                ClpTopNSpec topNSpec = topNSpecOptional.get();
+                // Only handles one range metadata column for now (first ordering)
+                ClpTopNSpec.Ordering ordering = topNSpec.getOrderings().get(0);
+                // Get the last column in the ordering (the primary sort column for nested fields)
+                String col = ordering.getColumns().get(ordering.getColumns().size() - 1);
+                String dir = (ordering.getOrder() == ClpTopNSpec.Order.ASC) ? "ASC" : "DESC";
+                String splitMetaQuery = format(SQL_SELECT_SPLIT_META_TEMPLATE, tableName, clpTableLayoutHandle.getMetadataSql().orElse("1 = 1"), col, dir);
+                List<ArchiveMeta> archiveMetaList = fetchArchiveMeta(splitMetaQuery, ordering);
+                List<ArchiveMeta> selected = selectTopNArchives(archiveMetaList, topNSpec.getLimit(), ordering.getOrder());
+
+                for (ArchiveMeta a : selected) {
+                    String splitPath = a.id;
+                    splits.add(new ClpSplit(splitPath, determineSplitType(splitPath), clpTableLayoutHandle.getKqlQuery()));
+                }
+
+                List<ClpSplit> filteredSplits = splits.build();
+                log.debug("Number of topN filtered splits: %s", filteredSplits.size());
+                return filteredSplits;
+            }
+
+            String splitQuery = format(SQL_SELECT_SPLITS_TEMPLATE, tableName, clpTableLayoutHandle.getMetadataSql().orElse("1 = 1"));
+            List<JsonNode> splitRows = getQueryResult(pinotDatabaseUrl, splitQuery);
+            for (JsonNode row : splitRows) {
+                String splitPath = row.elements().next().asText();
+                splits.add(new ClpSplit(splitPath, determineSplitType(splitPath), clpTableLayoutHandle.getKqlQuery()));
+            }
+
+            List<ClpSplit> filteredSplits = splits.build();
+            log.debug("Number of filtered splits: %s", filteredSplits.size());
+            return filteredSplits;
+        }
+        catch (Exception e) {
+            log.error(e, "Failed to list splits for table %s", tableName);
+            return Collections.emptyList();
+        }
+    }
+
+    /**
+     * Fetches archive metadata from the database.
+     *
+     * @param query    SQL query string that selects the archives
+     * @param ordering The top-N ordering specifying which columns contain lowerBound/upperBound
+     * @return List of ArchiveMeta objects representing archive metadata
+     */
+    private List<ArchiveMeta> fetchArchiveMeta(String query, ClpTopNSpec.Ordering ordering)
+    {
+        ImmutableList.Builder<ArchiveMeta> archiveMetas = new ImmutableList.Builder<>();
+        List<JsonNode> rows = getQueryResult(pinotDatabaseUrl, query);
+        for (JsonNode row : rows) {
+            archiveMetas.add(new ArchiveMeta(
+                    row.get(0).asText(),
+                    row.get(1).asLong(),
+                    row.get(2).asLong(),
+                    row.get(3).asLong()));
+        }
+        return archiveMetas.build();
+    }
+
+    /**
+     * Selects the set of archives that must be scanned to guarantee the top-N results by timestamp
+     * (ASC or DESC), given only archive ranges and message counts.
+     * <ul>
+     *   <li>Merges overlapping archives into components (union of time ranges).</li>
+     *   <li>For DESC: always include the newest component, then add older ones until their total
+     *      message counts cover the limit.</li>
+     *   <li>For ASC: symmetric — start from the oldest, then add newer ones.</li>
+     * </ul>
+
+     * @param archives list of archives with [lowerBound, upperBound, messageCount]
+     * @param limit number of messages requested
+     * @param order ASC (earliest first) or DESC (latest first)
+     * @return archives that must be scanned
+     */
+    private static List<ArchiveMeta> selectTopNArchives(List<ArchiveMeta> archives, long limit, ClpTopNSpec.Order order)
+    {
+        if (archives == null || archives.isEmpty() || limit <= 0) {
+            return ImmutableList.of();
+        }
+        requireNonNull(order, "order is null");
+
+        // 1) Merge overlaps into groups
+        List<ArchiveGroup> groups = toArchiveGroups(archives);
+
+        if (groups.isEmpty()) {
+            return ImmutableList.of();
+        }
+
+        // 2) Pick minimal set of groups per order, then return all member archives
+        List<ArchiveMeta> selected = new ArrayList<>();
+        if (order == ClpTopNSpec.Order.DESC) {
+            // newest group index
+            int k = groups.size() - 1;
+
+            // must include newest group
+            selected.addAll(groups.get(k).members);
+
+            // assume worst case: newest contributes 0 after filter; cover limit from older groups
+            long coveredByOlder = 0;
+            for (int i = k - 1; i >= 0 && coveredByOlder < limit; --i) {
+                selected.addAll(groups.get(i).members);
+                coveredByOlder += groups.get(i).count;
+            }
+        }
+        else {
+            // oldest group index
+            int k = 0;
+
+            // must include oldest group
+            selected.addAll(groups.get(k).members);
+
+            // assume worst case: oldest contributes 0; cover limit from newer groups
+            long coveredByNewer = 0;
+            for (int i = k + 1; i < groups.size() && coveredByNewer < limit; ++i) {
+                selected.addAll(groups.get(i).members);
+                coveredByNewer += groups.get(i).count;
+            }
+        }
+
+        return selected;
+    }
+
+    /**
+     * Groups overlapping archives into non-overlapping archive groups.
+     *
+     * @param archives archives sorted by lowerBound
+     * @return merged components
+     */
+    private static List<ArchiveGroup> toArchiveGroups(List<ArchiveMeta> archives)
+    {
+        List<ArchiveMeta> sorted = new ArrayList<>(archives);
+        sorted.sort(comparingLong((ArchiveMeta a) -> a.lowerBound)
+                .thenComparingLong(a -> a.upperBound));
+
+        List<ArchiveGroup> groups = new ArrayList<>();
+        ArchiveGroup cur = null;
+
+        for (ArchiveMeta a : sorted) {
+            if (cur == null) {
+                cur = startArchiveGroup(a);
+            }
+            else if (overlaps(cur, a)) {
+                // extend current component
+                cur.end = Math.max(cur.end, a.upperBound);
+                cur.count += a.messageCount;
+                cur.members.add(a);
+            }
+            else {
+                // finalize current, start a new one
+                groups.add(cur);
+                cur = startArchiveGroup(a);
+            }
+        }
+        if (cur != null) {
+            groups.add(cur);
+        }
+        return groups;
+    }
+
+    private static ArchiveGroup startArchiveGroup(ArchiveMeta a)
+    {
+        ArchiveGroup group = new ArchiveGroup();
+        group.begin = a.lowerBound;
+        group.end = a.upperBound;
+        group.count = a.messageCount;
+        group.members.add(a);
+        return group;
+    }
+
+    private static boolean overlaps(ArchiveGroup cur, ArchiveMeta a)
+    {
+        return a.lowerBound <= cur.end && a.upperBound >= cur.begin;
+    }
+
+    /**
+     * Determines the split type based on file path extension.
+     *
+     * @param splitPath the file path
+     * @return IR for .clp.zst files, ARCHIVE otherwise
+     */
+    private static SplitType determineSplitType(String splitPath)
+    {
+        return splitPath.endsWith(".clp.zst") ? IR : ARCHIVE;
+    }
+
+    private static List<JsonNode> getQueryResult(URL url, String sql)
+    {
+        try {
             HttpURLConnection conn = (HttpURLConnection) url.openConnection();
             conn.setRequestMethod("POST");
             conn.setRequestProperty("Content-Type", "application/json");
@@ -69,10 +272,9 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
             conn.setConnectTimeout((int) SECONDS.toMillis(5));
             conn.setReadTimeout((int) SECONDS.toMillis(30));
 
-            String query = format(SQL_SELECT_SPLITS_TEMPLATE, tableName, clpTableLayoutHandle.getMetadataSql().orElse("1 = 1"));
-            log.info("Pinot query: %s", query);
+            log.info("Executing Pinot query: %s", sql);
             ObjectMapper mapper = new ObjectMapper();
-            String body = format("{\"sql\": %s }", mapper.writeValueAsString(query));
+            String body = format("{\"sql\": %s }", mapper.writeValueAsString(sql));
             try (OutputStream os = conn.getOutputStream()) {
                 os.write(body.getBytes(StandardCharsets.UTF_8));
             }
@@ -89,26 +291,66 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
             }
             JsonNode resultTable = root.get("resultTable");
             if (resultTable == null) {
-                throw new RuntimeException("No \"resultTable\" field found");
+                throw new IllegalStateException("Pinot query response missing 'resultTable' field");
             }
             JsonNode rows = resultTable.get("rows");
             if (rows == null) {
-                throw new RuntimeException("No \"rows\" field found");
+                throw new IllegalStateException("Pinot query response missing 'rows' field in resultTable");
             }
+            ImmutableList.Builder<JsonNode> resultBuilder = ImmutableList.builder();
             for (Iterator<JsonNode> it = rows.elements(); it.hasNext(); ) {
                 JsonNode row = it.next();
-                String splitPath = row.elements().next().asText();
-                SplitType splitType = splitPath.endsWith(".clp.zst") ? IR : ARCHIVE;
-                splits.add(new ClpSplit(splitPath, splitType, clpTableLayoutHandle.getKqlQuery()));
+                resultBuilder.add(row);
             }
-            List<ClpSplit> filteredSplits = splits.build();
-            log.debug("Number of filtered splits: %s", filteredSplits.size());
-            return filteredSplits;
+            List<JsonNode> results = resultBuilder.build();
+            log.debug("Number of results: %s", results.size());
+            return results;
+        }
+        catch (IOException e) {
+            log.error(e, "IO error executing Pinot query: %s", sql);
+            return Collections.emptyList();
         }
         catch (Exception e) {
-            log.error(e);
+            log.error(e, "Unexpected error executing Pinot query: %s", sql);
+            return Collections.emptyList();
         }
+    }
+
+    /**
+     * Represents metadata of an archive, including its ID, timestamp bounds, and message count.
+     */
+    private static final class ArchiveMeta
+    {
+        private final String id;
+        private final long lowerBound;
+        private final long upperBound;
+        private final long messageCount;
 
-        return Collections.emptyList();
+        ArchiveMeta(String id, long lowerBound, long upperBound, long messageCount)
+        {
+            this.id = requireNonNull(id, "id is null");
+            if (lowerBound > upperBound) {
+                throw new IllegalArgumentException(
+                        format("Invalid archive bounds: lowerBound (%d) > upperBound (%d)", lowerBound, upperBound));
+            }
+            if (messageCount < 0) {
+                throw new IllegalArgumentException(
+                        format("Invalid message count: %d (must be >= 0)", messageCount));
+            }
+            this.lowerBound = lowerBound;
+            this.upperBound = upperBound;
+            this.messageCount = messageCount;
+        }
+    }
+
+    /**
+     * Represents a group of overlapping archives treated as one logical unit.
+     */
+    private static final class ArchiveGroup
+    {
+        long begin;
+        long end;
+        long count;
+        final List<ArchiveMeta> members = new ArrayList<>();
     }
 }