From d6541d94ab7050fee7742b85be74b2401a8968a1 Mon Sep 17 00:00:00 2001 From: wraymo Date: Mon, 22 Sep 2025 16:06:36 -0400 Subject: [PATCH 1/9] WIP feat: Add TopN optimizer for timestamp-based queries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement a plan optimizer that pushes down LIMIT + ORDER BY operations to reduce split scanning when querying on timestamp metadata columns. The optimizer detects TopN patterns in query plans and uses archive metadata (timestamp bounds, message counts) to select only the minimum set of archives needed to guarantee correct results. Key features: - Detects and rewrites TopN → [Project] → [Filter] → TableScan patterns - Pushes TopN specifications through ClpTableLayoutHandle - Handles overlapping archives by merging into non-overlapping groups - Supports both ASC (earliest-first) and DESC (latest-first) ordering - Uses worst-case analysis to ensure correctness when archive ranges overlap Implementation: - ClpTopNSpec: Data structure for TopN specifications (limit + orderings) - ClpComputePushDown: Enhanced plan optimizer with TopN detection - ClpMySqlSplitProvider: Archive selection logic using metadata queries - Comprehensive test coverage for various TopN scenarios Performance: For queries like "SELECT * FROM logs ORDER BY timestamp DESC LIMIT 100", this eliminates scanning of archives that cannot contain the top results, significantly reducing I/O. --- .../facebook/presto/plugin/clp/ClpSplit.java | 20 + .../plugin/clp/ClpTableLayoutHandle.java | 44 +- .../clp/optimization/ClpComputePushDown.java | 319 ++++++++++++- .../clp/{ => optimization}/ClpExpression.java | 51 +- .../optimization/ClpFilterToKqlConverter.java | 60 ++- .../ClpPlanOptimizerProvider.java | 5 +- .../plugin/clp/optimization/ClpTopNSpec.java | 144 ++++++ .../clp/split/ClpMySqlSplitProvider.java | 194 +++++++- .../filter/ClpMySqlSplitFilterProvider.java | 25 + .../split/filter/ClpSplitFilterProvider.java | 22 +- .../presto/plugin/clp/ClpMetadataDbSetUp.java | 22 +- .../presto/plugin/clp/TestClpFilterToKql.java | 1 + .../presto/plugin/clp/TestClpTopN.java | 440 ++++++++++++++++++ .../TestClpSplitFilterConfigCommon.java | 8 +- .../resources/test-topn-split-filter.json | 14 + 15 files changed, 1316 insertions(+), 53 deletions(-) rename presto-clp/src/main/java/com/facebook/presto/plugin/clp/{ => optimization}/ClpExpression.java (65%) create mode 100644 presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java create mode 100644 presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpTopN.java create mode 100644 presto-clp/src/test/resources/test-topn-split-filter.json diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java index 2e35840971c11..7b2d42bb0635d 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE; @@ -77,6 +78,25 @@ public List getPreferredNodes(NodeProvider nodeProvider) return ImmutableList.of(); } + @Override + public int hashCode() + { + return Objects.hash(path, type, kqlQuery); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + ClpSplit other = (ClpSplit) obj; + return this.type == other.type && this.path.equals(other.path) && this.kqlQuery.equals(other.kqlQuery); + } + @Override public Map getInfo() { diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java index b82932f0c30fd..902c9dfe37176 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java @@ -13,6 +13,7 @@ */ package com.facebook.presto.plugin.clp; +import com.facebook.presto.plugin.clp.optimization.ClpTopNSpec; import com.facebook.presto.spi.ConnectorTableLayoutHandle; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; @@ -28,13 +29,34 @@ public class ClpTableLayoutHandle private final ClpTableHandle table; private final Optional kqlQuery; private final Optional metadataSql; + private final boolean metadataQueryOnly; + private final Optional topN; @JsonCreator - public ClpTableLayoutHandle(@JsonProperty("table") ClpTableHandle table, @JsonProperty("kqlQuery") Optional kqlQuery, @JsonProperty("metadataFilterQuery") Optional metadataSql) + public ClpTableLayoutHandle( + @JsonProperty("table") ClpTableHandle table, + @JsonProperty("kqlQuery") Optional kqlQuery, + @JsonProperty("metadataFilterQuery") Optional metadataSql, + @JsonProperty("metadataQueryOnly") boolean metadataQueryOnly, + @JsonProperty("topN") Optional topN) { this.table = table; this.kqlQuery = kqlQuery; this.metadataSql = metadataSql; + this.metadataQueryOnly = metadataQueryOnly; + this.topN = topN; + } + + public ClpTableLayoutHandle( + @JsonProperty("table") ClpTableHandle table, + @JsonProperty("kqlQuery") Optional kqlQuery, + @JsonProperty("metadataFilterQuery") Optional metadataSql) + { + this.table = table; + this.kqlQuery = kqlQuery; + this.metadataSql = metadataSql; + this.metadataQueryOnly = false; + this.topN = Optional.empty(); } @JsonProperty @@ -55,6 +77,18 @@ public Optional getMetadataSql() return metadataSql; } + @JsonProperty + public boolean isMetadataQueryOnly() + { + return metadataQueryOnly; + } + + @JsonProperty + public Optional getTopN() + { + return topN; + } + @Override public boolean equals(Object o) { @@ -67,13 +101,15 @@ public boolean equals(Object o) ClpTableLayoutHandle that = (ClpTableLayoutHandle) o; return Objects.equals(table, that.table) && Objects.equals(kqlQuery, that.kqlQuery) && - Objects.equals(metadataSql, that.metadataSql); + Objects.equals(metadataSql, that.metadataSql) && + Objects.equals(metadataQueryOnly, that.metadataQueryOnly) && + Objects.equals(topN, that.topN); } @Override public int hashCode() { - return Objects.hash(table, kqlQuery, metadataSql); + return Objects.hash(table, kqlQuery, metadataSql, metadataQueryOnly, topN); } @Override @@ -83,6 +119,8 @@ public String toString() .add("table", table) .add("kqlQuery", kqlQuery) .add("metadataSql", metadataSql) + .add("metadataQueryOnly", metadataQueryOnly) + .add("topN", topN) .toString(); } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java index 2c216614af10f..c86ea0dbe4d7e 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java @@ -14,7 +14,10 @@ package com.facebook.presto.plugin.clp.optimization; import com.facebook.airlift.log.Logger; -import com.facebook.presto.plugin.clp.ClpExpression; +import com.facebook.presto.common.block.SortOrder; +import com.facebook.presto.common.type.RowType; +import com.facebook.presto.plugin.clp.ClpColumnHandle; +import com.facebook.presto.plugin.clp.ClpMetadata; import com.facebook.presto.plugin.clp.ClpTableHandle; import com.facebook.presto.plugin.clp.ClpTableLayoutHandle; import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider; @@ -22,25 +25,38 @@ import com.facebook.presto.spi.ConnectorPlanOptimizer; import com.facebook.presto.spi.ConnectorPlanRewriter; import com.facebook.presto.spi.ConnectorSession; +import com.facebook.presto.spi.ConnectorTableLayoutHandle; import com.facebook.presto.spi.TableHandle; import com.facebook.presto.spi.VariableAllocator; import com.facebook.presto.spi.function.FunctionMetadataManager; import com.facebook.presto.spi.function.StandardFunctionResolution; import com.facebook.presto.spi.plan.FilterNode; +import com.facebook.presto.spi.plan.Ordering; import com.facebook.presto.spi.plan.PlanNode; import com.facebook.presto.spi.plan.PlanNodeIdAllocator; +import com.facebook.presto.spi.plan.ProjectNode; import com.facebook.presto.spi.plan.TableScanNode; +import com.facebook.presto.spi.plan.TopNNode; +import com.facebook.presto.spi.relation.ConstantExpression; import com.facebook.presto.spi.relation.RowExpression; +import com.facebook.presto.spi.relation.SpecialFormExpression; import com.facebook.presto.spi.relation.VariableReferenceExpression; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import static com.facebook.presto.plugin.clp.ClpConnectorFactory.CONNECTOR_NAME; import static com.facebook.presto.spi.ConnectorPlanRewriter.rewriteWith; +import static java.lang.Math.toIntExact; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -67,7 +83,7 @@ public PlanNode optimize(PlanNode maxSubplan, ConnectorSession session, Variable // Throw exception if any required split filters are missing if (!rewriter.tableScopeSet.isEmpty() && !rewriter.hasVisitedFilter) { - splitFilterProvider.checkContainsRequiredFilters(rewriter.tableScopeSet, ""); + splitFilterProvider.checkContainsRequiredFilters(rewriter.tableScopeSet, ImmutableSet.of()); } return optimizedPlanNode; } @@ -105,6 +121,156 @@ public PlanNode visitFilter(FilterNode node, RewriteContext context) return processFilter(node, (TableScanNode) node.getSource()); } + @Override + public PlanNode visitTopN(TopNNode node, RewriteContext context) + { + PlanNode rewrittenSource = context.rewrite(node.getSource(), null); + + ProjectNode project = null; + FilterNode filter = null; + PlanNode cursor = rewrittenSource; + + if (cursor instanceof ProjectNode) { + project = (ProjectNode) cursor; + cursor = project.getSource(); + } + if (cursor instanceof FilterNode) { + filter = (FilterNode) cursor; + cursor = filter.getSource(); + } + if (!(cursor instanceof TableScanNode)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + TableScanNode scan = (TableScanNode) cursor; + TableHandle tableHandle = scan.getTable(); + if (!(tableHandle.getConnectorHandle() instanceof ClpTableHandle)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // only allow TopN pushdown when metadata-only is true + boolean metadataOnly = false; + Optional layout = tableHandle.getLayout(); + Optional kql = Optional.empty(); + Optional metadataSql = Optional.empty(); + Optional existingTopN = Optional.empty(); + ClpTableHandle clpTableHandle = null; + + if (layout.isPresent() && layout.get() instanceof ClpTableLayoutHandle) { + ClpTableLayoutHandle cl = (ClpTableLayoutHandle) layout.get(); + metadataOnly = cl.isMetadataQueryOnly(); + kql = cl.getKqlQuery(); + metadataSql = cl.getMetadataSql(); + existingTopN = cl.getTopN(); + clpTableHandle = cl.getTable(); + } + + if (!metadataOnly) { + // Rule: skip TopN pushdown unless metadataQueryOnly is true + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // Ensure ORDER BY items are plain variables (allow identity through Project) + List ords = node.getOrderingScheme().getOrderBy(); + if (project != null && !areIdents(project, ords)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + Map assignments = scan.getAssignments(); + List newOrderings = new ArrayList<>(ords.size()); + for (Ordering ord : ords) { + VariableReferenceExpression outVar = ord.getVariable(); + Optional columnNameOpt = buildOrderColumnName(project, outVar, assignments); + if (!columnNameOpt.isPresent()) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + String tableScope = CONNECTOR_NAME + "." + (clpTableHandle != null ? + clpTableHandle.getSchemaTableName().toString() : ClpMetadata.DEFAULT_SCHEMA_NAME); + + List remappedColumnName = splitFilterProvider.remapColumnName(tableScope, columnNameOpt.get()); + newOrderings.add(new ClpTopNSpec.Ordering(remappedColumnName, toClpOrder(ord.getSortOrder()))); + } + + if (existingTopN.isPresent()) { + ClpTopNSpec ex = existingTopN.get(); + if (!sameOrdering(ex.getOrderings(), newOrderings)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); // leave existing as-is + } + long mergedLimit = Math.min(ex.getLimit(), node.getCount()); + if (mergedLimit == ex.getLimit()) { + // No change needed; keep current layout/spec + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // Tighten the limit on the layout + ClpTopNSpec tightened = new ClpTopNSpec(mergedLimit, ex.getOrderings()); + ClpTableHandle clpHandle = (ClpTableHandle) tableHandle.getConnectorHandle(); + ClpTableLayoutHandle newLayout = + new ClpTableLayoutHandle(clpHandle, kql, metadataSql, true, Optional.of(tightened)); + + TableScanNode newScan = new TableScanNode( + scan.getSourceLocation(), + idAllocator.getNextId(), + new TableHandle( + tableHandle.getConnectorId(), + clpHandle, + tableHandle.getTransaction(), + Optional.of(newLayout)), + scan.getOutputVariables(), + scan.getAssignments(), + scan.getTableConstraints(), + scan.getCurrentConstraint(), + scan.getEnforcedConstraint(), + scan.getCteMaterializationInfo()); + + PlanNode newSource = newScan; + if (filter != null) { + newSource = new FilterNode(filter.getSourceLocation(), idAllocator.getNextId(), newSource, filter.getPredicate()); + } + if (project != null) { + newSource = new ProjectNode( + project.getSourceLocation(), + idAllocator.getNextId(), + newSource, + project.getAssignments(), + project.getLocality()); + } + + return new TopNNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, node.getCount(), node.getOrderingScheme(), node.getStep()); + } + + ClpTopNSpec spec = new ClpTopNSpec(node.getCount(), newOrderings); + ClpTableHandle clpHandle = (ClpTableHandle) tableHandle.getConnectorHandle(); + ClpTableLayoutHandle newLayout = + new ClpTableLayoutHandle(clpHandle, kql, metadataSql, true, Optional.of(spec)); + + TableScanNode newScanNode = new TableScanNode( + scan.getSourceLocation(), + idAllocator.getNextId(), + new TableHandle( + tableHandle.getConnectorId(), + clpHandle, + tableHandle.getTransaction(), + Optional.of(newLayout)), + scan.getOutputVariables(), + scan.getAssignments(), + scan.getTableConstraints(), + scan.getCurrentConstraint(), + scan.getEnforcedConstraint(), + scan.getCteMaterializationInfo()); + + PlanNode newSource = newScanNode; + if (filter != null) { + newSource = new FilterNode(filter.getSourceLocation(), idAllocator.getNextId(), newSource, filter.getPredicate()); + } + if (project != null) { + newSource = new ProjectNode(project.getSourceLocation(), idAllocator.getNextId(), newSource, project.getAssignments(), project.getLocality()); + } + + return new TopNNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, node.getCount(), node.getOrderingScheme(), node.getStep()); + } + private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNode) { hasVisitedFilter = true; @@ -114,21 +280,23 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod String tableScope = CONNECTOR_NAME + "." + clpTableHandle.getSchemaTableName().toString(); Map assignments = tableScanNode.getAssignments(); + Set metadataColumnNames = splitFilterProvider.getColumnNames(tableScope); ClpExpression clpExpression = filterNode.getPredicate().accept( new ClpFilterToKqlConverter( functionResolution, functionManager, assignments, - splitFilterProvider.getColumnNames(tableScope)), + metadataColumnNames), null); + Optional kqlQuery = clpExpression.getPushDownExpression(); Optional metadataSqlQuery = clpExpression.getMetadataSqlQuery(); Optional remainingPredicate = clpExpression.getRemainingExpression(); // Perform required metadata filter checks before handling the KQL query (if kqlQuery // isn't present, we'll return early, skipping subsequent checks). - splitFilterProvider.checkContainsRequiredFilters(ImmutableSet.of(tableScope), metadataSqlQuery.orElse("")); + splitFilterProvider.checkContainsRequiredFilters(ImmutableSet.of(tableScope), clpExpression.getPushDownVariables()); boolean hasMetadataFilter = metadataSqlQuery.isPresent() && !metadataSqlQuery.get().isEmpty(); if (hasMetadataFilter) { metadataSqlQuery = Optional.of(splitFilterProvider.remapSplitFilterPushDownExpression(tableScope, metadataSqlQuery.get())); @@ -140,7 +308,12 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod log.debug("KQL query: %s", kqlQuery.get()); } - ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle(clpTableHandle, kqlQuery, metadataSqlQuery); + ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle( + clpTableHandle, + kqlQuery, + metadataSqlQuery, + metadataColumnNames.equals(clpExpression.getPushDownVariables()), + Optional.empty()); TableHandle newTableHandle = new TableHandle( tableHandle.getConnectorId(), clpTableHandle, @@ -171,5 +344,141 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod return tableScanNode; } } + + private boolean sameOrdering(List a, List b) + { + if (a.size() != b.size()) { + return false; + } + for (int i = 0; i < a.size(); i++) { + ClpTopNSpec.Ordering x = a.get(i); + ClpTopNSpec.Ordering y = b.get(i); + if (!Objects.equals(x.getColumns(), y.getColumns())) { + return false; + } + if (x.getOrder() != y.getOrder()) { + return false; + } + } + return true; + } + + /** Accept plain var or dereference-of-var passthroughs. */ + private boolean areIdents(ProjectNode project, List vars) + { + for (Ordering ord : vars) { + VariableReferenceExpression out = ord.getVariable(); + RowExpression expr = project.getAssignments().get(out); + + if (expr instanceof VariableReferenceExpression) { + continue; + } + if (isDereferenceChainOverVariable(expr)) { + continue; + } + return false; + } + return true; + } + + /** Build final column name string for CLP (e.g., "msg.timestamp"), or empty if not pushdownable. */ + private Optional buildOrderColumnName( + ProjectNode project, + VariableReferenceExpression outVar, + Map assignments) + { + if (project == null) { + // ORDER BY directly on scan var + ColumnHandle ch = assignments.get(outVar); + if (!(ch instanceof ClpColumnHandle)) { + return Optional.empty(); + } + return Optional.of(((ClpColumnHandle) ch).getOriginalColumnName()); + } + + RowExpression expr = project.getAssignments().get(outVar); + if (expr instanceof VariableReferenceExpression) { + ColumnHandle ch = assignments.get((VariableReferenceExpression) expr); + if (!(ch instanceof ClpColumnHandle)) { + return Optional.empty(); + } + return Optional.of(((ClpColumnHandle) ch).getOriginalColumnName()); + } + + // Handle DEREFERENCE chain: baseVar.field1.field2... + Deque path = new ArrayDeque<>(); + RowExpression cur = expr; + + while (cur instanceof SpecialFormExpression + && ((SpecialFormExpression) cur).getForm() == SpecialFormExpression.Form.DEREFERENCE) { + SpecialFormExpression s = (SpecialFormExpression) cur; + RowExpression base = s.getArguments().get(0); + RowExpression indexExpr = s.getArguments().get(1); + + if (!(indexExpr instanceof ConstantExpression) || !(base.getType() instanceof RowType)) { + return Optional.empty(); + } + int idx; + Object v = ((ConstantExpression) indexExpr).getValue(); + if (v instanceof Long) { + idx = toIntExact((Long) v); + } + else if (v instanceof Integer) { + idx = (Integer) v; + } + else { + return Optional.empty(); + } + + RowType rowType = (RowType) base.getType(); + if (idx < 0 || idx >= rowType.getFields().size()) { + return Optional.empty(); + } + String fname = rowType.getFields().get(idx).getName().orElse(String.valueOf(idx)); + // We traverse outer->inner; collect in deque and join later + path.addLast(fname); + + cur = base; // move up the chain + } + + if (!(cur instanceof VariableReferenceExpression)) { + return Optional.empty(); + } + + ColumnHandle baseCh = assignments.get((VariableReferenceExpression) cur); + if (!(baseCh instanceof ClpColumnHandle)) { + return Optional.empty(); + } + + String baseName = ((ClpColumnHandle) baseCh).getOriginalColumnName(); + if (path.isEmpty()) { + return Optional.of(baseName); + } + return Optional.of(baseName + "." + String.join(".", path)); + } + + /** True if expr is DEREFERENCE(... DEREFERENCE(baseVar, i) ..., j) with baseVar a VariableReferenceExpression. */ + private boolean isDereferenceChainOverVariable(RowExpression expr) + { + RowExpression cur = expr; + while (cur instanceof SpecialFormExpression + && ((SpecialFormExpression) cur).getForm() == SpecialFormExpression.Form.DEREFERENCE) { + cur = ((SpecialFormExpression) cur).getArguments().get(0); + } + return (cur instanceof VariableReferenceExpression); + } + + private ClpTopNSpec.Order toClpOrder(SortOrder so) + { + switch (so) { + case ASC_NULLS_FIRST: + case ASC_NULLS_LAST: + return ClpTopNSpec.Order.ASC; + case DESC_NULLS_FIRST: + case DESC_NULLS_LAST: + return ClpTopNSpec.Order.DESC; + default: throw new IllegalArgumentException("Unknown sort order: " + so); + } + } } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java similarity index 65% rename from presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java rename to presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java index e970f9848a9cf..571ecb028dc0a 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java @@ -11,11 +11,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.facebook.presto.plugin.clp; +package com.facebook.presto.plugin.clp.optimization; import com.facebook.presto.spi.relation.RowExpression; +import com.google.common.collect.ImmutableSet; import java.util.Optional; +import java.util.Set; /** * Represents the result of: @@ -38,11 +40,19 @@ public class ClpExpression // The remaining (non-translatable) portion of the RowExpression, if any. private final Optional remainingExpression; - public ClpExpression(String pushDownExpression, String metadataSqlQuery, RowExpression remainingExpression) + // Variables used in pushDownExpression + private final Set pushDownVariables; + + public ClpExpression( + String pushDownExpression, + String metadataSqlQuery, + RowExpression remainingExpression, + Set pushDownVariables) { this.pushDownExpression = Optional.ofNullable(pushDownExpression); this.metadataSqlQuery = Optional.ofNullable(metadataSqlQuery); this.remainingExpression = Optional.ofNullable(remainingExpression); + this.pushDownVariables = ImmutableSet.copyOf(pushDownVariables); } /** @@ -50,7 +60,7 @@ public ClpExpression(String pushDownExpression, String metadataSqlQuery, RowExpr */ public ClpExpression() { - this(null, null, null); + this(null, null, null, ImmutableSet.of()); } /** @@ -60,7 +70,18 @@ public ClpExpression() */ public ClpExpression(String pushDownExpression) { - this(pushDownExpression, null, null); + this(pushDownExpression, null, null, ImmutableSet.of()); + } + + /** + * Creates a ClpExpression from a fully translatable KQL query or column name. + * + * @param pushDownExpression + * @param pushDownVariables + */ + public ClpExpression(String pushDownExpression, Set pushDownVariables) + { + this(pushDownExpression, null, null, pushDownVariables); } /** @@ -72,7 +93,20 @@ public ClpExpression(String pushDownExpression) */ public ClpExpression(String pushDownExpression, String metadataSqlQuery) { - this(pushDownExpression, metadataSqlQuery, null); + this(pushDownExpression, metadataSqlQuery, null, ImmutableSet.of()); + } + + /** + * Creates a ClpExpression from a fully translatable KQL string or column name, as well as a + * metadata SQL string. + * + * @param pushDownExpression + * @param metadataSqlQuery + * @param pushDownVariables + */ + public ClpExpression(String pushDownExpression, String metadataSqlQuery, Set pushDownVariables) + { + this(pushDownExpression, metadataSqlQuery, null, pushDownVariables); } /** @@ -82,7 +116,7 @@ public ClpExpression(String pushDownExpression, String metadataSqlQuery) */ public ClpExpression(RowExpression remainingExpression) { - this(null, null, remainingExpression); + this(null, null, remainingExpression, ImmutableSet.of()); } public Optional getPushDownExpression() @@ -99,4 +133,9 @@ public Optional getRemainingExpression() { return remainingExpression; } + + public Set getPushDownVariables() + { + return pushDownVariables; + } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java index b27a61ef0d65a..df6658664c349 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java @@ -19,7 +19,6 @@ import com.facebook.presto.common.type.Type; import com.facebook.presto.common.type.VarcharType; import com.facebook.presto.plugin.clp.ClpColumnHandle; -import com.facebook.presto.plugin.clp.ClpExpression; import com.facebook.presto.spi.ColumnHandle; import com.facebook.presto.spi.PrestoException; import com.facebook.presto.spi.function.FunctionHandle; @@ -163,7 +162,8 @@ public ClpExpression visitConstant(ConstantExpression node, Void context) @Override public ClpExpression visitVariableReference(VariableReferenceExpression node, Void context) { - return new ClpExpression(getVariableName(node)); + String variableName = getVariableName(node); + return new ClpExpression(variableName, ImmutableSet.of(variableName)); } @Override @@ -250,7 +250,8 @@ private ClpExpression handleBetween(CallExpression node) return new ClpExpression(node); } - Optional variableOpt = first.accept(this, null).getPushDownExpression(); + ClpExpression variableExpression = first.accept(this, null); + Optional variableOpt = variableExpression.getPushDownExpression(); if (!variableOpt.isPresent() || !(second instanceof ConstantExpression) || !(third instanceof ConstantExpression)) { @@ -264,7 +265,7 @@ private ClpExpression handleBetween(CallExpression node) String metadataSqlQuery = metadataFilterColumns.contains(variable) ? String.format("\"%s\" >= %s AND \"%s\" <= %s", variable, lowerBound, variable, upperBound) : null; - return new ClpExpression(kql, metadataSqlQuery); + return new ClpExpression(kql, metadataSqlQuery, variableExpression.getPushDownVariables()); } /** @@ -290,10 +291,10 @@ private ClpExpression handleNot(CallExpression node) } String notPushDownExpression = "NOT " + expression.getPushDownExpression().get(); if (expression.getMetadataSqlQuery().isPresent()) { - return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSqlQuery()); + return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSqlQuery(), expression.getPushDownVariables()); } else { - return new ClpExpression(notPushDownExpression); + return new ClpExpression(notPushDownExpression, expression.getPushDownVariables()); } } @@ -345,7 +346,7 @@ else if (argument instanceof CallExpression) { return new ClpExpression(node); } pattern = pattern.replace("%", "*").replace("_", "?"); - return new ClpExpression(format("%s: \"%s\"", variableName, pattern)); + return new ClpExpression(format("%s: \"%s\"", variableName, pattern), variable.getPushDownVariables()); } /** @@ -444,31 +445,36 @@ private ClpExpression buildClpExpression( String metadataSqlQuery = null; if (operator.equals(EQUAL)) { if (literalType instanceof VarcharType) { - return new ClpExpression(format("%s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString))); + return new ClpExpression( + format("%s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), + ImmutableSet.of(variableName)); } else { if (metadataFilterColumns.contains(variableName)) { metadataSqlQuery = format("\"%s\" = %s", variableName, literalString); } - return new ClpExpression(format("%s: %s", variableName, literalString), metadataSqlQuery); + return new ClpExpression(format("%s: %s", variableName, literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } } else if (operator.equals(NOT_EQUAL)) { if (literalType instanceof VarcharType) { - return new ClpExpression(format("NOT %s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString))); + return new ClpExpression( + format("NOT %s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), + ImmutableSet.of(variableName)); } else { if (metadataFilterColumns.contains(variableName)) { metadataSqlQuery = format("NOT \"%s\" = %s", variableName, literalString); } - return new ClpExpression(format("NOT %s: %s", variableName, literalString), metadataSqlQuery); + return new ClpExpression(format("NOT %s: %s", variableName, literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } } else if (LOGICAL_BINARY_OPS_FILTER.contains(operator) && !(literalType instanceof VarcharType)) { if (metadataFilterColumns.contains(variableName)) { - metadataSqlQuery = format("\"%s\" %s %s", variableName, operator.getOperator(), literalString); + metadataSqlQuery = format("\"%s\" %s %s", variableName, operator.getOperator(), literalString, ImmutableSet.of(variableName)); } - return new ClpExpression(format("%s %s %s", variableName, operator.getOperator(), literalString), metadataSqlQuery); + return new ClpExpression( + format("%s %s %s", variableName, operator.getOperator(), literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } return new ClpExpression(originalNode); } @@ -576,7 +582,7 @@ private Optional interpretSubstringEquality(SubstrInfo info, Stri result.append("?"); } result.append(targetString).append("*\""); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } } } @@ -590,11 +596,11 @@ private Optional interpretSubstringEquality(SubstrInfo info, Stri result.append("?"); } result.append(targetString).append("\""); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } if (start == -targetString.length()) { result.append(format("%s: \"*%s\"", info.variableName, targetString)); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } } } @@ -678,10 +684,12 @@ private ClpExpression handleAnd(SpecialFormExpression node) List remainingExpressions = new ArrayList<>(); boolean hasMetadataSql = false; boolean hasPushDownExpression = false; + ImmutableSet.Builder pushDownVariables = new ImmutableSet.Builder<>(); for (RowExpression argument : node.getArguments()) { ClpExpression expression = argument.accept(this, null); if (expression.getPushDownExpression().isPresent()) { hasPushDownExpression = true; + pushDownVariables.addAll(expression.getPushDownVariables()); queryBuilder.append(expression.getPushDownExpression().get()); queryBuilder.append(" AND "); if (expression.getMetadataSqlQuery().isPresent()) { @@ -702,18 +710,21 @@ else if (!remainingExpressions.isEmpty()) { return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 5) + ")", hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, - remainingExpressions.get(0)); + remainingExpressions.get(0), + pushDownVariables.build()); } else { return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 5) + ")", hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, - new SpecialFormExpression(node.getSourceLocation(), AND, BOOLEAN, remainingExpressions)); + new SpecialFormExpression(node.getSourceLocation(), AND, BOOLEAN, remainingExpressions), + pushDownVariables.build()); } } // Remove the last " AND " from the query return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 5) + ")", - hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null); + hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, + pushDownVariables.build()); } /** @@ -736,6 +747,7 @@ private ClpExpression handleOr(SpecialFormExpression node) queryBuilder.append("("); boolean allPushedDown = true; boolean hasAllMetadataSql = true; + ImmutableSet.Builder pushDownVariables = new ImmutableSet.Builder<>(); for (RowExpression argument : node.getArguments()) { ClpExpression expression = argument.accept(this, null); // Note: It is possible in the future that an expression cannot be pushed down as a KQL query, but can be @@ -746,6 +758,7 @@ private ClpExpression handleOr(SpecialFormExpression node) } queryBuilder.append(expression.getPushDownExpression().get()); queryBuilder.append(" OR "); + pushDownVariables.addAll(expression.getPushDownVariables()); if (hasAllMetadataSql && expression.getMetadataSqlQuery().isPresent()) { metadataQueryBuilder.append(expression.getMetadataSqlQuery().get()); metadataQueryBuilder.append(" OR "); @@ -758,7 +771,8 @@ private ClpExpression handleOr(SpecialFormExpression node) // Remove the last " OR " from the query return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 4) + ")", - hasAllMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 4) + ")" : null); + hasAllMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 4) + ")" : null, + pushDownVariables.build()); } return new ClpExpression(node); } @@ -798,7 +812,7 @@ private ClpExpression handleIn(SpecialFormExpression node) } // Remove the last " OR " from the query - return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 4) + ")"); + return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 4) + ")", variable.getPushDownVariables()); } /** @@ -823,7 +837,7 @@ private ClpExpression handleIsNull(SpecialFormExpression node) } String variableName = expression.getPushDownExpression().get(); - return new ClpExpression(format("NOT %s: *", variableName)); + return new ClpExpression(format("NOT %s: *", variableName), expression.getPushDownVariables()); } /** @@ -885,7 +899,7 @@ private ClpExpression handleDereference(RowExpression expression) if (!baseString.getPushDownExpression().isPresent()) { return new ClpExpression(expression); } - return new ClpExpression(baseString.getPushDownExpression().get() + "." + fieldName); + return new ClpExpression(baseString.getPushDownExpression().get() + "." + fieldName, baseString.getPushDownVariables()); } /** diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java index b536c95ad216a..bdf50eb0fb709 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java @@ -32,7 +32,10 @@ public class ClpPlanOptimizerProvider private final ClpSplitFilterProvider splitFilterProvider; @Inject - public ClpPlanOptimizerProvider(FunctionMetadataManager functionManager, StandardFunctionResolution functionResolution, ClpSplitFilterProvider splitFilterProvider) + public ClpPlanOptimizerProvider( + FunctionMetadataManager functionManager, + StandardFunctionResolution functionResolution, + ClpSplitFilterProvider splitFilterProvider) { this.functionManager = functionManager; this.functionResolution = functionResolution; diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java new file mode 100644 index 0000000000000..fa5f06852aed7 --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java @@ -0,0 +1,144 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.facebook.presto.plugin.clp.optimization; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.List; +import java.util.Objects; + +import static java.util.Objects.requireNonNull; + +/** + * Represents a Top-N specification for a query, including the limit of rows to return + * and the ordering of columns. + */ +public class ClpTopNSpec +{ + /** + * Enum representing the order direction: ascending or descending. + */ + public enum Order + { + ASC, + DESC + } + + /** + * Represents the ordering of one or more columns with a specified order (ASC or DESC). + */ + public static final class Ordering + { + private final List columns; + private final Order order; + + @JsonCreator + public Ordering(@JsonProperty("columns") List columns, @JsonProperty("order") Order order) + { + this.columns = requireNonNull(columns, "column is null"); + this.order = requireNonNull(order, "order is null"); + } + + @JsonProperty + public List getColumns() + { + return columns; + } + + @JsonProperty + public Order getOrder() + { + return order; + } + + @Override + public int hashCode() + { + return Objects.hash(columns, order); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + Ordering other = (Ordering) obj; + return this.order == other.order && this.columns.equals(other.columns); + } + + @Override + public String toString() + { + return columns + ":" + order; + } + } + + private final long limit; + private final List orderings; + + @JsonCreator + public ClpTopNSpec(@JsonProperty("limit") long limit, @JsonProperty("orderings") List orderings) + { + if (limit <= 0) { + throw new IllegalArgumentException("limit must be > 0"); + } + if (orderings == null || orderings.isEmpty()) { + throw new IllegalArgumentException("orderings must be non-empty"); + } + this.limit = limit; + this.orderings = orderings; + } + + @JsonProperty + public long getLimit() + { + return limit; + } + + @JsonProperty + public List getOrderings() + { + return orderings; + } + + @Override + public int hashCode() + { + return Objects.hash(limit, orderings); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + ClpTopNSpec other = (ClpTopNSpec) obj; + return this.limit == other.limit && this.orderings.equals(other.orderings); + } + + @Override + public String toString() + { + return "ClpTopNSpec (limit=" + limit + ", order=" + orderings + ")"; + } +} diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java index 6b54218509c7f..13435e28a8b2c 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.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.google.common.collect.ImmutableList; import javax.inject.Inject; @@ -27,22 +28,26 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; import java.util.List; +import java.util.Optional; import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.ARCHIVE; import static java.lang.String.format; +import static java.util.Comparator.comparingLong; public class ClpMySqlSplitProvider implements ClpSplitProvider { // Column names public static final String ARCHIVES_TABLE_COLUMN_ID = "id"; + public static final String ARCHIVES_TABLE_NUM_MESSAGES = "num_messages"; // Table suffixes public static final String ARCHIVES_TABLE_SUFFIX = "_archives"; // SQL templates - private static final String SQL_SELECT_ARCHIVES_TEMPLATE = format("SELECT `%s` FROM `%%s%%s%s` WHERE 1 = 1", ARCHIVES_TABLE_COLUMN_ID, ARCHIVES_TABLE_SUFFIX); + private static final String SQL_SELECT_ARCHIVES_TEMPLATE = format("SELECT * FROM `%%s%%s%s` WHERE 1 = 1", ARCHIVES_TABLE_SUFFIX); private static final Logger log = Logger.get(ClpMySqlSplitProvider.class); @@ -66,6 +71,7 @@ public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) { ImmutableList.Builder splits = new ImmutableList.Builder<>(); ClpTableHandle clpTableHandle = clpTableLayoutHandle.getTable(); + Optional topNSpecOptional = clpTableLayoutHandle.getTopN(); String tablePath = clpTableHandle.getTablePath(); String tableName = clpTableHandle.getSchemaTableName().getTableName(); String archivePathQuery = format(SQL_SELECT_ARCHIVES_TEMPLATE, config.getMetadataTablePrefix(), tableName); @@ -74,6 +80,25 @@ public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) String metadataFilterQuery = clpTableLayoutHandle.getMetadataSql().get(); archivePathQuery += " AND (" + metadataFilterQuery + ")"; } + + if (topNSpecOptional.isPresent()) { + ClpTopNSpec topNSpec = topNSpecOptional.get(); + // Only handles one range metadata column for now + ClpTopNSpec.Ordering ordering = topNSpec.getOrderings().get(0); + String col = ordering.getColumns().get(ordering.getColumns().size() - 1); + String dir = (ordering.getOrder() == ClpTopNSpec.Order.ASC) ? "ASC" : "DESC"; + archivePathQuery += " ORDER BY " + "`" + col + "` " + dir; + + List archiveMetaList = fetchArchiveMeta(archivePathQuery, ordering); + List selected = selectTopNArchives(archiveMetaList, topNSpec.getLimit(), ordering.getOrder()); + + for (ArchiveMeta a : selected) { + splits.add(new ClpSplit(tablePath + "/" + a.id, ARCHIVE, clpTableLayoutHandle.getKqlQuery())); + } + ImmutableList result = splits.build(); + log.debug("Number of splits: %s", result.size()); + return result; + } log.debug("Query for archive: %s", archivePathQuery); try (Connection connection = getConnection()) { @@ -105,4 +130,171 @@ private Connection getConnection() } return connection; } + + /** + * 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 fetchArchiveMeta(String query, ClpTopNSpec.Ordering ordering) + { + List list = new ArrayList<>(); + try (Connection connection = getConnection(); + PreparedStatement stmt = connection.prepareStatement(query); + ResultSet rs = stmt.executeQuery()) { + while (rs.next()) { + list.add(new ArchiveMeta( + rs.getString(ARCHIVES_TABLE_COLUMN_ID), + rs.getLong(ordering.getColumns().get(0)), + rs.getLong(ordering.getColumns().get(1)), + rs.getLong(ARCHIVES_TABLE_NUM_MESSAGES))); + } + } + catch (SQLException e) { + log.warn("Database error while fetching archive metadata: %s", e); + } + return list; + } + + /** + * 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. + *
    + *
  • Merges overlapping archives into groups (union of time ranges).
    • + *
  • For DESC: always include the newest group, then add older ones until their total + * message counts cover the limit.
    • + *
  • For ASC: symmetric — start from the oldest, then add newer ones.
    • + *
+ + * @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 selectTopNArchives(List archives, long limit, ClpTopNSpec.Order order) + { + if (archives == null || archives.isEmpty() || limit <= 0) { + return ImmutableList.of(); + } + + // 1) Merge overlaps into groups + List groups = toArchiveGroups(archives); + + // 2) Pick minimal set of groups per order, then return all member archives + List 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 groups + */ + private static List toArchiveGroups(List archives) + { + List sorted = new ArrayList<>(archives); + sorted.sort(comparingLong((ArchiveMeta a) -> a.lowerBound) + .thenComparingLong(a -> a.upperBound)); + + List groups = new ArrayList<>(); + ArchiveGroup cur = null; + + for (ArchiveMeta a : sorted) { + if (cur == null) { + cur = startArchiveGroup(a); + } + else if (overlaps(cur, a)) { + // extend current group + 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; + } + + /** + * Represents metadata of an archive, including its ID, timestamp bounds, and message count. + */ + private static class ArchiveMeta + { + final String id; + final long lowerBound; + final long upperBound; + final long messageCount; + + ArchiveMeta(String id, long lowerBound, long upperBound, long messageCount) + { + this.id = id; + 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 members = new ArrayList<>(); + } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java index 31d24fd4df71c..4bec8a79c9eed 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java @@ -15,6 +15,7 @@ import com.facebook.presto.plugin.clp.ClpConfig; import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; @@ -85,6 +86,30 @@ public String remapSplitFilterPushDownExpression(String scope, String pushDownEx return remappedSql; } + @Override + public List remapColumnName(String scope, String columnName) + { + String[] splitScope = scope.split("\\."); + + Map mappings = new HashMap<>(getAllMappingsFromFilters(filterMap.get(splitScope[0]))); + + if (1 < splitScope.length) { + mappings.putAll(getAllMappingsFromFilters(filterMap.get(splitScope[0] + "." + splitScope[1]))); + } + + if (3 == splitScope.length) { + mappings.putAll(getAllMappingsFromFilters(filterMap.get(scope))); + } + + if (mappings.containsKey(columnName)) { + ClpMySqlCustomSplitFilterOptions.RangeMapping value = mappings.get(columnName); + return ImmutableList.of(value.lowerBound, value.upperBound); + } + else { + return ImmutableList.of(columnName); + } + } + @Override protected Class getCustomSplitFilterOptionsClass() { diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java index 0609843aaf22f..7f19a5296b801 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java @@ -91,20 +91,36 @@ public ClpSplitFilterProvider(ClpConfig config) */ public abstract String remapSplitFilterPushDownExpression(String scope, String pushDownExpression); + /** + * Rewrites {@code columnName} to remap column names based on the {@code "customOptions"} for + * the given scope. + *

+ * {@code scope} follows the format {@code catalog[.schema][.table]}, and determines which + * column mappings to apply, since mappings from more specific scopes (e.g., table-level) + * override or supplement those from broader scopes (e.g., catalog-level). For each scope + * (catalog, schema, table), this method collects all mappings defined in + * {@code "customOptions"}. + * + * @param scope the scope of the column mapping + * @param columnName the column name to be remapped + * @return the remapped column names + */ + public abstract List remapColumnName(String scope, String columnName); + /** * Checks for the given table, if {@code splitFilterPushDownExpression} contains all required * fields. * * @param tableScopeSet the set of scopes of the tables that are being queried - * @param splitFilterPushDownExpression the expression to be checked + * @param pushDownVariables the set of variables being pushed down */ - public void checkContainsRequiredFilters(Set tableScopeSet, String splitFilterPushDownExpression) + public void checkContainsRequiredFilters(Set tableScopeSet, Set pushDownVariables) { boolean hasRequiredSplitFilterColumns = true; ImmutableList.Builder notFoundListBuilder = ImmutableList.builder(); for (String tableScope : tableScopeSet) { for (String columnName : getRequiredColumnNames(tableScope)) { - if (!splitFilterPushDownExpression.contains(columnName)) { + if (!pushDownVariables.contains(columnName)) { hasRequiredSplitFilterColumns = false; notFoundListBuilder.add(columnName); } diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java index d1d0ee6964c8e..ee207f9864004 100644 --- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java @@ -37,6 +37,7 @@ import static com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider.DATASETS_TABLE_COLUMN_NAME; import static com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider.DATASETS_TABLE_SUFFIX; import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_COLUMN_ID; +import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_NUM_MESSAGES; import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_SUFFIX; import static java.lang.String.format; import static java.util.UUID.randomUUID; @@ -139,26 +140,30 @@ public static ClpMySqlSplitProvider setupSplit(DbHandle dbHandle, Map("msg.timestamp", Integer), + new Pair<>("city.Name", ClpString), + new Pair<>("city.Region.Id", Integer), + new Pair<>("city.Region.Name", VarString), + new Pair<>("fare", Float), + new Pair<>("isHoliday", Boolean)))); + + splitProvider = setupSplit(dbHandle, + ImmutableMap.of( + tableName, + ImmutableList.of( + new ClpMetadataDbSetUp.ArchivesTableRow("0", 100, 0, 100), + new ClpMetadataDbSetUp.ArchivesTableRow("1", 100, 50, 150), + new ClpMetadataDbSetUp.ArchivesTableRow("2", 100, 100, 200), + new ClpMetadataDbSetUp.ArchivesTableRow("3", 100, 201, 300), + new ClpMetadataDbSetUp.ArchivesTableRow("4", 100, 301, 400)))); + + URL resource = getClass().getClassLoader().getResource("test-topn-split-filter.json"); + if (resource == null) { + log.error("test-topn-split-filter.json not found in resources"); + return; + } + + String filterConfigPath; + try { + filterConfigPath = Paths.get(resource.toURI()).toAbsolutePath().toString(); + } + catch (URISyntaxException e) { + log.error("test-topn-split-filter.json not found in resources"); + return; + } + + localQueryRunner = new LocalQueryRunner(defaultSession); + localQueryRunner.createCatalog("clp", new ClpConnectorFactory(), ImmutableMap.of( + "clp.metadata-db-url", format(METADATA_DB_URL_TEMPLATE, dbHandle.getDbPath()), + "clp.metadata-db-user", METADATA_DB_USER, + "clp.metadata-db-password", METADATA_DB_PASSWORD, + "clp.metadata-table-prefix", METADATA_DB_TABLE_PREFIX)); + localQueryRunner.getMetadata().registerBuiltInFunctions(extractFunctions(new ClpPlugin().getFunctions())); + functionAndTypeManager = localQueryRunner.getMetadata().getFunctionAndTypeManager(); + functionResolution = new FunctionResolution(functionAndTypeManager.getFunctionAndTypeResolver()); + splitFilterProvider = new ClpMySqlSplitFilterProvider(new ClpConfig().setSplitFilterConfig(filterConfigPath)); + planNodeIdAllocator = new PlanNodeIdAllocator(); + variableAllocator = new VariableAllocator(); + } + + @AfterMethod + public void tearDown() + { + localQueryRunner.close(); + ClpMetadataDbSetUp.tearDown(dbHandle); + } + + @Test + public void test() + { + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 120 AND msg.timestamp < 240 ORDER BY msg.timestamp DESC LIMIT 100", + "(msg.timestamp > 120 AND msg.timestamp < 240)", + "(end_timestamp > 120 AND begin_timestamp < 240)", + 100, + DESC, + ImmutableSet.of("1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 120 AND msg.timestamp < 240 ORDER BY msg.timestamp ASC LIMIT 50", + "(msg.timestamp > 120 AND msg.timestamp < 240)", + "(end_timestamp > 120 AND begin_timestamp < 240)", + 50, + ASC, + ImmutableSet.of("1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp >= 180 AND msg.timestamp <= 260 ORDER BY msg.timestamp DESC LIMIT 100", + "(msg.timestamp >= 180 AND msg.timestamp <= 260)", + "(end_timestamp >= 180 AND begin_timestamp <= 260)", + 100, + DESC, + ImmutableSet.of("2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 250 AND msg.timestamp < 290 ORDER BY msg.timestamp DESC LIMIT 10", + "(msg.timestamp > 250 AND msg.timestamp < 290)", + "(end_timestamp > 250 AND begin_timestamp < 290)", + 10, + DESC, + ImmutableSet.of("3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 1000 AND msg.timestamp < 1100 ORDER BY msg.timestamp DESC LIMIT 10", + "(msg.timestamp > 1000 AND msg.timestamp < 1100)", + "(end_timestamp > 1000 AND begin_timestamp < 1100)", + 10, + DESC, + ImmutableSet.of()); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp <= 300 ORDER BY msg.timestamp DESC LIMIT 1000", + "msg.timestamp <= 300", + "begin_timestamp <= 300", + 1000, + DESC, + ImmutableSet.of("0", "1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp <= 400 ORDER BY msg.timestamp DESC LIMIT 100", + "msg.timestamp <= 400", + "begin_timestamp <= 400", + 100, + DESC, + ImmutableSet.of("3", "4")); + } + + private void testTopNQueryPlanAndSplits(String sql, String kql, String metadataSql, long limit, Order order, Set splitIds) + { + TransactionId transactionId = localQueryRunner.getTransactionManager().beginTransaction(false); + Session session = testSessionBuilder().setCatalog("clp").setSchema("default").setTransactionId(transactionId).build(); + + Plan plan = localQueryRunner.createPlan( + session, + sql, + WarningCollector.NOOP); + ClpComputePushDown optimizer = new ClpComputePushDown(functionAndTypeManager, functionResolution, splitFilterProvider); + PlanNode optimizedPlan = optimizer.optimize(plan.getRoot(), session.toConnectorSession(), variableAllocator, planNodeIdAllocator); + PlanNode optimizedPlanWithUniqueId = freshenIds(optimizedPlan, new PlanNodeIdAllocator()); + + ClpTableLayoutHandle clpTableLayoutHandle = new ClpTableLayoutHandle( + table, + Optional.of(kql), + Optional.of(metadataSql), + true, + Optional.of(new ClpTopNSpec( + limit, + ImmutableList.of(new ClpTopNSpec.Ordering(ImmutableList.of("begin_timestamp", "end_timestamp"), order))))); + + PlanAssert.assertPlan( + session, + localQueryRunner.getMetadata(), + (node, sourceStats, lookup, s, types) -> PlanNodeStatsEstimate.unknown(), + new Plan(optimizedPlanWithUniqueId, plan.getTypes(), StatsAndCosts.empty()), + anyTree( + ClpTableScanMatcher.clpTableScanPattern( + clpTableLayoutHandle, + ImmutableSet.of( + city, + fare, + isHoliday, + new ClpColumnHandle( + "msg", + RowType.from(ImmutableList.of(new RowType.Field(Optional.of("timestamp"), BIGINT)))))))); + + assertEquals( + ImmutableSet.copyOf(splitProvider.listSplits(clpTableLayoutHandle)), + splitIds.stream() + .map(id -> new ClpSplit("/tmp/archives/test/" + id, ARCHIVE, Optional.of(kql))) + .collect(ImmutableSet.toImmutableSet())); + } + + /** + * Recursively rebuilds a query plan tree so that every {@link PlanNode} has a fresh, unique ID. + *

+ * This utility is mainly for testing, to avoid ID collisions that can occur when + * localQueryRunner.createPlan() and a custom optimizer each use separate + * {@link PlanNodeIdAllocator}s that start at the same seed, producing duplicate IDs. + * + * @param root the root of the plan + * @param idAlloc the plan node ID allocator + * @return the plan with a fresh, unique IDs. + */ + private static PlanNode freshenIds(PlanNode root, PlanNodeIdAllocator idAlloc) + { + return SimplePlanRewriter.rewriteWith(new SimplePlanRewriter() { + @Override + public PlanNode visitOutput(OutputNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new OutputNode( + node.getSourceLocation(), + idAlloc.getNextId(), + src, + node.getColumnNames(), + node.getOutputVariables()); + } + + @Override + public PlanNode visitExchange(ExchangeNode node, RewriteContext ctx) + { + List newSources = node.getSources().stream() + .map(s -> ctx.rewrite(s, null)) + .collect(com.google.common.collect.ImmutableList.toImmutableList()); + + return new ExchangeNode( + node.getSourceLocation(), + idAlloc.getNextId(), + node.getType(), + node.getScope(), + node.getPartitioningScheme(), + newSources, + node.getInputs(), + node.isEnsureSourceOrdering(), + node.getOrderingScheme()); + } + + @Override + public PlanNode visitProject(ProjectNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new ProjectNode(idAlloc.getNextId(), src, node.getAssignments()); + } + + @Override + public PlanNode visitFilter(FilterNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new FilterNode(node.getSourceLocation(), idAlloc.getNextId(), src, node.getPredicate()); + } + + @Override + public PlanNode visitTopN(TopNNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new TopNNode( + node.getSourceLocation(), + idAlloc.getNextId(), + src, + node.getCount(), + node.getOrderingScheme(), + node.getStep()); + } + + @Override + public PlanNode visitTableScan(TableScanNode node, RewriteContext ctx) + { + return new TableScanNode( + node.getSourceLocation(), + idAlloc.getNextId(), + node.getTable(), + node.getOutputVariables(), + node.getAssignments()); + } + + @Override + public PlanNode visitPlan(PlanNode node, RewriteContext ctx) + { + List newChildren = node.getSources().stream() + .map(ch -> ctx.rewrite(ch, null)) + .collect(com.google.common.collect.ImmutableList.toImmutableList()); + return node.replaceChildren(newChildren); + } + }, root, null); + } + + private static final class ClpTableScanMatcher + implements Matcher + { + private final ClpTableLayoutHandle expectedLayoutHandle; + private final Set expectedColumns; + + private ClpTableScanMatcher(ClpTableLayoutHandle expectedLayoutHandle, Set expectedColumns) + { + this.expectedLayoutHandle = expectedLayoutHandle; + this.expectedColumns = expectedColumns; + } + + static PlanMatchPattern clpTableScanPattern(ClpTableLayoutHandle layoutHandle, Set columns) + { + return node(TableScanNode.class).with(new ClpTableScanMatcher(layoutHandle, columns)); + } + + @Override + public boolean shapeMatches(PlanNode node) + { + return node instanceof TableScanNode; + } + + @Override + public MatchResult detailMatches( + PlanNode node, + StatsProvider stats, + Session session, + Metadata metadata, + SymbolAliases symbolAliases) + { + checkState(shapeMatches(node), "Plan testing framework error: shapeMatches returned false"); + TableScanNode tableScanNode = (TableScanNode) node; + ClpTableLayoutHandle actualLayoutHandle = (ClpTableLayoutHandle) tableScanNode.getTable().getLayout().get(); + + // Check layout handle + if (!expectedLayoutHandle.equals(actualLayoutHandle)) { + return NO_MATCH; + } + + // Check assignments contain expected columns + Map actualAssignments = tableScanNode.getAssignments(); + Set actualColumns = new HashSet<>(actualAssignments.values()); + + if (!expectedColumns.equals(actualColumns)) { + return NO_MATCH; + } + + SymbolAliases.Builder aliasesBuilder = SymbolAliases.builder(); + for (VariableReferenceExpression variable : tableScanNode.getOutputVariables()) { + aliasesBuilder.put(variable.getName(), new SymbolReference(variable.getName())); + } + + return match(aliasesBuilder.build()); + } + } +} diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java index 7a4058f617d0c..c6fe1cee8bec2 100644 --- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java @@ -57,12 +57,8 @@ public void checkRequiredFilters() config.setSplitFilterConfig(filterConfigPath); ClpMySqlSplitFilterProvider filterProvider = new ClpMySqlSplitFilterProvider(config); Set testTableScopeSet = ImmutableSet.of(format("%s.%s", CONNECTOR_NAME, new SchemaTableName("default", "table_1"))); - assertThrows(PrestoException.class, () -> filterProvider.checkContainsRequiredFilters( - testTableScopeSet, - "(\"level\" >= 1 AND \"level\" <= 3)")); - filterProvider.checkContainsRequiredFilters( - testTableScopeSet, - "(\"msg.timestamp\" > 1234 AND \"msg.timestamp\" < 5678)"); + assertThrows(PrestoException.class, () -> filterProvider.checkContainsRequiredFilters(testTableScopeSet, ImmutableSet.of("level"))); + filterProvider.checkContainsRequiredFilters(testTableScopeSet, ImmutableSet.of("msg.timestamp")); } @Test diff --git a/presto-clp/src/test/resources/test-topn-split-filter.json b/presto-clp/src/test/resources/test-topn-split-filter.json new file mode 100644 index 0000000000000..53450716cb7b4 --- /dev/null +++ b/presto-clp/src/test/resources/test-topn-split-filter.json @@ -0,0 +1,14 @@ +{ + "clp.default.test": [ + { + "columnName": "msg.timestamp", + "customOptions": { + "rangeMapping": { + "lowerBound": "begin_timestamp", + "upperBound": "end_timestamp" + } + }, + "required": true + } + ] +} From 86c0d06ea3f56a27eeb68cee20594585aca130b2 Mon Sep 17 00:00:00 2001 From: Jack Luo Date: Fri, 19 Sep 2025 12:42:53 +0000 Subject: [PATCH 2/9] WIP feat: Add YAML metadata provider and Pinot split provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add alternative provider implementations for metadata and split discovery, enabling flexible deployment configurations without MySQL dependency. New providers: - ClpYamlMetadataProvider: Reads table/column metadata from YAML config files instead of querying MySQL database. Supports nested schemas and polymorphic types through hierarchical YAML structure. - ClpPinotSplitProvider: Queries Apache Pinot database for archive file paths via HTTP/JSON API instead of MySQL. Determines split types based on file extensions (.clp.zst for IR, otherwise ARCHIVE). Configuration: - Add clp.metadata-provider-type=YAML option for YAML-based metadata - Add clp.split-provider-type=PINOT option for Pinot-based split discovery - Add clp.metadata-yaml-path config for YAML metadata file location The YAML metadata provider parses a two-level structure: 1. Main metadata file: maps connector → schema → table → schema_file_path 2. Individual table schema files: define column names and types This provides deployment flexibility - users can now choose between: - Database-driven (MySQL for both metadata and splits) - Hybrid (YAML metadata + Pinot splits) - Or any combination based on their infrastructure Dependencies: Adds jackson-dataformat-yaml and snakeyaml 2.1 --- presto-clp/pom.xml | 15 ++ .../facebook/presto/plugin/clp/ClpConfig.java | 20 ++- .../presto/plugin/clp/ClpErrorCode.java | 1 + .../facebook/presto/plugin/clp/ClpModule.java | 8 + .../clp/metadata/ClpYamlMetadataProvider.java | 140 ++++++++++++++++++ .../clp/split/ClpPinotSplitProvider.java | 114 ++++++++++++++ .../plugin/clp/TestClpYamlMetadata.java | 129 ++++++++++++++++ presto-native-execution/pom.xml | 10 ++ 8 files changed, 435 insertions(+), 2 deletions(-) create mode 100644 presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java create mode 100644 presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java create mode 100644 presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java diff --git a/presto-clp/pom.xml b/presto-clp/pom.xml index d13b59968aff5..8941da517cecb 100644 --- a/presto-clp/pom.xml +++ b/presto-clp/pom.xml @@ -75,6 +75,11 @@ jackson-databind + + com.fasterxml.jackson.dataformat + jackson-dataformat-yaml + + com.facebook.presto presto-spi @@ -154,4 +159,14 @@ test + + + + + org.yaml + snakeyaml + 2.1 + + + diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java index 121eb0d5ff17b..329d5451c71a5 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java @@ -33,6 +33,8 @@ public class ClpConfig private long metadataRefreshInterval = 60; private long metadataExpireInterval = 600; + private String metadataYamlPath; + private String splitFilterConfig; private SplitFilterProviderType splitFilterProviderType = SplitFilterProviderType.MYSQL; private SplitProviderType splitProviderType = SplitProviderType.MYSQL; @@ -151,6 +153,18 @@ public ClpConfig setMetadataExpireInterval(long metadataExpireInterval) return this; } + public String getMetadataYamlPath() + { + return metadataYamlPath; + } + + @Config("clp.metadata-yaml-path") + public ClpConfig setMetadataYamlPath(String metadataYamlPath) + { + this.metadataYamlPath = metadataYamlPath; + return this; + } + public String getSplitFilterConfig() { return splitFilterConfig; @@ -189,7 +203,8 @@ public ClpConfig setSplitProviderType(SplitProviderType splitProviderType) public enum MetadataProviderType { - MYSQL + MYSQL, + YAML } public enum SplitFilterProviderType @@ -199,6 +214,7 @@ public enum SplitFilterProviderType public enum SplitProviderType { - MYSQL + MYSQL, + PINOT } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java index 2530c013455cc..fb6626de25a61 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java @@ -28,6 +28,7 @@ public enum ClpErrorCode CLP_UNSUPPORTED_SPLIT_SOURCE(2, EXTERNAL), CLP_UNSUPPORTED_TYPE(3, EXTERNAL), CLP_UNSUPPORTED_CONFIG_OPTION(4, EXTERNAL), + CLP_UNSUPPORTED_TABLE_SCHEMA_YAML(5, EXTERNAL), CLP_SPLIT_FILTER_CONFIG_NOT_FOUND(10, USER_ERROR), CLP_MANDATORY_SPLIT_FILTER_NOT_VALID(11, USER_ERROR), diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java index bf801d0d87242..6d979ca0bceb8 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java @@ -16,7 +16,9 @@ import com.facebook.airlift.configuration.AbstractConfigurationAwareModule; import com.facebook.presto.plugin.clp.metadata.ClpMetadataProvider; import com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider; +import com.facebook.presto.plugin.clp.metadata.ClpYamlMetadataProvider; import com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider; +import com.facebook.presto.plugin.clp.split.ClpPinotSplitProvider; import com.facebook.presto.plugin.clp.split.ClpSplitProvider; import com.facebook.presto.plugin.clp.split.filter.ClpMySqlSplitFilterProvider; import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider; @@ -56,6 +58,9 @@ protected void setup(Binder binder) if (config.getMetadataProviderType() == MetadataProviderType.MYSQL) { binder.bind(ClpMetadataProvider.class).to(ClpMySqlMetadataProvider.class).in(Scopes.SINGLETON); } + else if (config.getMetadataProviderType() == MetadataProviderType.YAML) { + binder.bind(ClpMetadataProvider.class).to(ClpYamlMetadataProvider.class).in(Scopes.SINGLETON); + } else { throw new PrestoException(CLP_UNSUPPORTED_METADATA_SOURCE, "Unsupported metadata provider type: " + config.getMetadataProviderType()); } @@ -63,6 +68,9 @@ protected void setup(Binder binder) if (config.getSplitProviderType() == SplitProviderType.MYSQL) { binder.bind(ClpSplitProvider.class).to(ClpMySqlSplitProvider.class).in(Scopes.SINGLETON); } + else if (config.getSplitProviderType() == SplitProviderType.PINOT) { + binder.bind(ClpSplitProvider.class).to(ClpPinotSplitProvider.class).in(Scopes.SINGLETON); + } else { throw new PrestoException(CLP_UNSUPPORTED_SPLIT_SOURCE, "Unsupported split provider type: " + config.getSplitProviderType()); } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java new file mode 100644 index 0000000000000..0bbb0b4c9ea79 --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java @@ -0,0 +1,140 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.facebook.presto.plugin.clp.metadata; + +import com.facebook.airlift.log.Logger; +import com.facebook.presto.plugin.clp.ClpColumnHandle; +import com.facebook.presto.plugin.clp.ClpConfig; +import com.facebook.presto.plugin.clp.ClpTableHandle; +import com.facebook.presto.spi.PrestoException; +import com.facebook.presto.spi.SchemaTableName; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + +import javax.inject.Inject; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static com.facebook.presto.plugin.clp.ClpConnectorFactory.CONNECTOR_NAME; +import static com.facebook.presto.plugin.clp.ClpErrorCode.CLP_UNSUPPORTED_TABLE_SCHEMA_YAML; +import static java.lang.String.format; + +public class ClpYamlMetadataProvider + implements ClpMetadataProvider +{ + private static final Logger log = Logger.get(ClpYamlMetadataProvider.class); + private final ClpConfig config; + private Map tableSchemaYamlMap; + + @Inject + public ClpYamlMetadataProvider(ClpConfig config) + { + this.config = config; + } + + @Override + public List listColumnHandles(SchemaTableName schemaTableName) + { + Path tableSchemaPath = Paths.get(tableSchemaYamlMap.get(schemaTableName)); + ClpSchemaTree schemaTree = new ClpSchemaTree(config.isPolymorphicTypeEnabled()); + ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); + + try { + Map root = mapper.readValue( + new File(tableSchemaPath.toString()), + new TypeReference>() {}); + ImmutableList.Builder namesBuilder = ImmutableList.builder(); + ImmutableList.Builder typesBuilder = ImmutableList.builder(); + collectTypes(root, "", namesBuilder, typesBuilder); + ImmutableList names = namesBuilder.build(); + ImmutableList types = typesBuilder.build(); + // The names and types should have same sizes + for (int i = 0; i < names.size(); i++) { + schemaTree.addColumn(names.get(i), types.get(i)); + } + return schemaTree.collectColumnHandles(); + } + catch (IOException e) { + log.error(format("Failed to parse table schema file %s, error: %s", tableSchemaPath, e.getMessage()), e); + } + return Collections.emptyList(); + } + + @Override + public List listTableHandles(String schemaName) + { + Path tablesSchemaPath = Paths.get(config.getMetadataYamlPath()); + ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); + + try { + Map root = mapper.readValue(new File(tablesSchemaPath.toString()), + new TypeReference>() {}); + + Object catalogObj = root.get(CONNECTOR_NAME); + if (!(catalogObj instanceof Map)) { + throw new PrestoException(CLP_UNSUPPORTED_TABLE_SCHEMA_YAML, format("The table schema does not contain field: %s", CONNECTOR_NAME)); + } + Object schemaObj = ((Map) catalogObj).get(schemaName); + ImmutableList.Builder tableHandlesBuilder = new ImmutableList.Builder<>(); + ImmutableMap.Builder tableSchemaYamlMapBuilder = new ImmutableMap.Builder<>(); + for (Map.Entry schemaEntry : ((Map) schemaObj).entrySet()) { + String tableName = schemaEntry.getKey(); + String tableSchemaYamlPath = schemaEntry.getValue().toString(); + // The splits' absolute paths will be stored in Pinot metadata database + SchemaTableName schemaTableName = new SchemaTableName(schemaName, tableName); + tableHandlesBuilder.add(new ClpTableHandle(schemaTableName, "")); + tableSchemaYamlMapBuilder.put(schemaTableName, tableSchemaYamlPath); + } + this.tableSchemaYamlMap = tableSchemaYamlMapBuilder.build(); + return tableHandlesBuilder.build(); + } + catch (IOException e) { + log.error(format("Failed to parse metadata file: %s, error: %s", config.getMetadataYamlPath(), e.getMessage()), e); + } + return Collections.emptyList(); + } + + private void collectTypes(Object node, String prefix, ImmutableList.Builder namesBuilder, ImmutableList.Builder typesBuilder) + { + if (node instanceof Number) { + namesBuilder.add(prefix); + typesBuilder.add(((Number) node).byteValue()); + return; + } + if (node instanceof List) { + for (Number type : (List) node) { + namesBuilder.add(prefix); + typesBuilder.add(type.byteValue()); + } + return; + } + for (Map.Entry entry : ((Map) node).entrySet()) { + if (!prefix.isEmpty()) { + collectTypes(entry.getValue(), format("%s.%s", prefix, entry.getKey()), namesBuilder, typesBuilder); + continue; + } + collectTypes(entry.getValue(), entry.getKey(), namesBuilder, typesBuilder); + } + } +} diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java new file mode 100644 index 0000000000000..45b471b96773e --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java @@ -0,0 +1,114 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.facebook.presto.plugin.clp.split; + +import com.facebook.airlift.log.Logger; +import com.facebook.presto.plugin.clp.ClpConfig; +import com.facebook.presto.plugin.clp.ClpSplit; +import com.facebook.presto.plugin.clp.ClpTableHandle; +import com.facebook.presto.plugin.clp.ClpTableLayoutHandle; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; + +import javax.inject.Inject; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; + +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.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 final ClpConfig config; + + @Inject + public ClpPinotSplitProvider(ClpConfig config) + { + this.config = config; + } + + @Override + public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) + { + ImmutableList.Builder splits = new ImmutableList.Builder<>(); + ClpTableHandle clpTableHandle = clpTableLayoutHandle.getTable(); + String tableName = clpTableHandle.getSchemaTableName().getTableName(); + try { + URL url = new URL(config.getMetadataDbUrl() + "/query/sql"); + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setRequestMethod("POST"); + conn.setRequestProperty("Content-Type", "application/json"); + conn.setRequestProperty("Accept", "application/json"); + conn.setDoOutput(true); + 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); + ObjectMapper mapper = new ObjectMapper(); + String body = format("{\"sql\": %s }", mapper.writeValueAsString(query)); + try (OutputStream os = conn.getOutputStream()) { + os.write(body.getBytes(StandardCharsets.UTF_8)); + } + + int code = conn.getResponseCode(); + InputStream is = (code >= 200 && code < 300) ? conn.getInputStream() : conn.getErrorStream(); + if (is == null) { + throw new IOException("Pinot HTTP " + code + " with empty body"); + } + + JsonNode root; + try (InputStream in = is) { + root = mapper.readTree(in); + } + JsonNode resultTable = root.get("resultTable"); + if (resultTable == null) { + throw new RuntimeException("No \"resultTable\" field found"); + } + JsonNode rows = resultTable.get("rows"); + if (rows == null) { + throw new RuntimeException("No \"rows\" field found"); + } + for (Iterator 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())); + } + List filteredSplits = splits.build(); + log.debug("Number of filtered splits: %s", filteredSplits.size()); + return filteredSplits; + } + catch (Exception e) { + log.error(e); + } + + return Collections.emptyList(); + } +} diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java new file mode 100644 index 0000000000000..e5a45d05692ae --- /dev/null +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java @@ -0,0 +1,129 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.facebook.presto.plugin.clp; + +import com.facebook.presto.plugin.clp.metadata.ClpMetadataProvider; +import com.facebook.presto.plugin.clp.metadata.ClpYamlMetadataProvider; +import com.facebook.presto.plugin.clp.split.ClpPinotSplitProvider; +import com.facebook.presto.plugin.clp.split.ClpSplitProvider; +import com.facebook.presto.spi.ColumnMetadata; +import com.facebook.presto.spi.ConnectorTableMetadata; +import com.facebook.presto.spi.SchemaTableName; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import java.util.HashSet; +import java.util.List; +import java.util.Optional; + +import static com.facebook.presto.plugin.clp.ClpConfig.MetadataProviderType.YAML; +import static com.facebook.presto.plugin.clp.ClpMetadata.DEFAULT_SCHEMA_NAME; +import static com.facebook.presto.testing.TestingConnectorSession.SESSION; +import static org.testng.Assert.assertEquals; + +public class TestClpYamlMetadata +{ + private static final String PINOT_BROKER_URL = "http://localhost:8099"; + private static final String METADATA_YAML_PATH = "/home/xiaochong-dev/presto-e2e/pinot/tables-schema.yaml"; + private static final String TABLE_NAME = "cockroachdb"; + private ClpMetadata metadata; + private ClpSplitProvider clpSplitProvider; + + @BeforeTest + public void setUp() + { + ClpConfig config = new ClpConfig() + .setPolymorphicTypeEnabled(true) + .setMetadataDbUrl(PINOT_BROKER_URL) + .setMetadataProviderType(YAML) + .setMetadataYamlPath(METADATA_YAML_PATH); + ClpMetadataProvider metadataProvider = new ClpYamlMetadataProvider(config); + metadata = new ClpMetadata(config, metadataProvider); + clpSplitProvider = new ClpPinotSplitProvider(config); + } + + @Test + public void testListSchemaNames() + { + assertEquals(metadata.listSchemaNames(SESSION), ImmutableList.of(DEFAULT_SCHEMA_NAME)); + } + + @Test + public void testListTables() + { + ImmutableSet.Builder builder = ImmutableSet.builder(); + builder.add(new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME)); + assertEquals(new HashSet<>(metadata.listTables(SESSION, Optional.empty())), builder.build()); + } + + @Test + public void testListSplits() + { + ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle( + new ClpTableHandle(new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME), ""), + Optional.empty(), + Optional.empty()); + List result = clpSplitProvider.listSplits(layoutHandle); + System.out.println("Hello world"); + } + + @Test + public void testGetTableMetadata() + { + ClpTableHandle clpTableHandle = (ClpTableHandle) metadata.getTableHandle(SESSION, new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME)); + ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(SESSION, clpTableHandle); +// ImmutableSet columnMetadata = ImmutableSet.builder() +// .add(ColumnMetadata.builder() +// .setName("a_bigint") +// .setType(BIGINT) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("a_varchar") +// .setType(VARCHAR) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("b_double") +// .setType(DOUBLE) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("b_varchar") +// .setType(VARCHAR) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("c") +// .setType(RowType.from(ImmutableList.of( +// RowType.field("d", BOOLEAN), +// RowType.field("e", VARCHAR)))) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("f") +// .setType(RowType.from(ImmutableList.of( +// RowType.field("g", +// RowType.from(ImmutableList.of( +// RowType.field("h", new ArrayType(VARCHAR)))))))) +// .setNullable(true) +// .build()) +// .build(); +// assertEquals(columnMetadata, ImmutableSet.copyOf(tableMetadata.getColumns())); + ImmutableSet actual = ImmutableSet.copyOf(tableMetadata.getColumns()); + System.out.println("Hello world"); + } +} diff --git a/presto-native-execution/pom.xml b/presto-native-execution/pom.xml index 200ffa6834afc..adbd01ab4b917 100644 --- a/presto-native-execution/pom.xml +++ b/presto-native-execution/pom.xml @@ -267,6 +267,16 @@ + + + + org.yaml + snakeyaml + 2.1 + + + +