getPhysicalPlanOptimizers()
{
- return ImmutableSet.of(new ClpPlanOptimizer(functionManager, functionResolution, splitFilterProvider));
+ return ImmutableSet.of(new ClpComputePushDown(functionManager, functionResolution, splitFilterProvider));
}
}
diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpUdfRewriter.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpUdfRewriter.java
new file mode 100644
index 0000000000000..aff40041e2d58
--- /dev/null
+++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpUdfRewriter.java
@@ -0,0 +1,306 @@
+/*
+ * 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.facebook.presto.plugin.clp.ClpColumnHandle;
+import com.facebook.presto.spi.ColumnHandle;
+import com.facebook.presto.spi.ConnectorPlanOptimizer;
+import com.facebook.presto.spi.ConnectorPlanRewriter;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.VariableAllocator;
+import com.facebook.presto.spi.function.FunctionMetadataManager;
+import com.facebook.presto.spi.plan.Assignments;
+import com.facebook.presto.spi.plan.FilterNode;
+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.relation.CallExpression;
+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 io.airlift.slice.Slice;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static com.facebook.presto.plugin.clp.ClpErrorCode.CLP_PUSHDOWN_UNSUPPORTED_EXPRESSION;
+import static com.facebook.presto.spi.ConnectorPlanRewriter.rewriteWith;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Utility for rewriting CLP UDFs (e.g., CLP_GET_*) in {@link RowExpression} trees.
+ *
+ * Traverses a query plan and rewrites calls to CLP_GET_* UDFs into
+ * {@link VariableReferenceExpression}s with meaningful names derived from their arguments.
+ *
+ * This enables querying fields that are not part of the original table schema but are available
+ * in CLP.
+ */
+public final class ClpUdfRewriter
+ implements ConnectorPlanOptimizer
+{
+ private final FunctionMetadataManager functionManager;
+
+ public ClpUdfRewriter(FunctionMetadataManager functionManager)
+ {
+ this.functionManager = requireNonNull(functionManager, "functionManager is null");
+ }
+
+ @Override
+ public PlanNode optimize(PlanNode maxSubplan, ConnectorSession session, VariableAllocator allocator, PlanNodeIdAllocator idAllocator)
+ {
+ return rewriteWith(new Rewriter(idAllocator, allocator, collectExistingScanAssignments(maxSubplan)), maxSubplan);
+ }
+
+ /**
+ * Collects all existing variable assignments from {@link TableScanNode} instances that map
+ * {@link ColumnHandle}s to {@link VariableReferenceExpression}s.
+ *
+ * This method traverses the given plan subtree, visiting the {@link TableScanNode} and
+ * extracting its assignments. The resulting map allows tracking which scan-level variables
+ * already exist for specific columns, so that subsequent optimizer passes (e.g., CLP UDF
+ * rewriting) can reuse them instead of creating duplicates.
+ *
+ * The key of the returned map is the {@link ColumnHandle}, and the value is the
+ * {@link VariableReferenceExpression} assigned to it in the scan node.
+ *
+ * @param root the root {@link PlanNode} of the plan subtree
+ * @return a map from column handle to its correh hsponding scan-level variable
+ */
+ private Map collectExistingScanAssignments(PlanNode root)
+ {
+ Map map = new HashMap<>();
+ root.getSources().forEach(source -> map.putAll(collectExistingScanAssignments(source)));
+ if (root instanceof TableScanNode) {
+ TableScanNode scan = (TableScanNode) root;
+ scan.getAssignments().forEach((var, handle) -> {
+ map.putIfAbsent(handle, var);
+ });
+ }
+ return map;
+ }
+
+ private class Rewriter
+ extends ConnectorPlanRewriter
+ {
+ private final PlanNodeIdAllocator idAllocator;
+ private final VariableAllocator variableAllocator;
+ private final Map globalColumnVarMap;
+
+ public Rewriter(
+ PlanNodeIdAllocator idAllocator,
+ VariableAllocator variableAllocator,
+ Map globalColumnVarMap)
+ {
+ this.idAllocator = idAllocator;
+ this.variableAllocator = variableAllocator;
+ this.globalColumnVarMap = globalColumnVarMap;
+ }
+
+ @Override
+ public PlanNode visitProject(ProjectNode node, RewriteContext context)
+ {
+ Assignments.Builder newAssignments = Assignments.builder();
+ for (Map.Entry entry : node.getAssignments().getMap().entrySet()) {
+ newAssignments.put(
+ entry.getKey(),
+ rewriteClpUdfs(entry.getValue(), functionManager, variableAllocator));
+ }
+
+ PlanNode newSource = rewritePlanSubtree(node.getSource());
+ return new ProjectNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, newAssignments.build(), node.getLocality());
+ }
+
+ @Override
+ public PlanNode visitFilter(FilterNode node, RewriteContext context)
+ {
+ return buildNewFilterNode(node);
+ }
+
+ /**
+ * Rewrites CLP_GET_* UDFs in a {@link RowExpression}, collecting each
+ * resulting variable into the given map along with its associated {@link ColumnHandle}.
+ *
+ * Each CLP_GET_* UDF must take a single constant string argument, which is
+ * used to construct the name of the variable reference (e.g.
+ * CLP_GET_STRING('foo') becomes a variable name foo). Invalid
+ * usages (e.g., non-constant arguments) will throw a {@link PrestoException}.
+ *
+ * @param expression the input expression to analyze and possibly rewrite
+ * @param functionManager function manager used to resolve function metadata
+ * @param variableAllocator variable allocator used to create new variable references
+ * @return a possibly rewritten {@link RowExpression} with CLP_GET_* calls
+ * replaced
+ */
+ private RowExpression rewriteClpUdfs(
+ RowExpression expression,
+ FunctionMetadataManager functionManager,
+ VariableAllocator variableAllocator)
+ {
+ // Handle CLP_GET_* function calls
+ if (expression instanceof CallExpression) {
+ CallExpression call = (CallExpression) expression;
+ String functionName = functionManager.getFunctionMetadata(call.getFunctionHandle()).getName().getObjectName().toUpperCase();
+
+ if (functionName.startsWith("CLP_GET_")) {
+ if (call.getArguments().size() != 1 || !(call.getArguments().get(0) instanceof ConstantExpression)) {
+ throw new PrestoException(CLP_PUSHDOWN_UNSUPPORTED_EXPRESSION,
+ "CLP_GET_* UDF must have a single constant string argument");
+ }
+
+ ConstantExpression constant = (ConstantExpression) call.getArguments().get(0);
+ String jsonPath = ((Slice) constant.getValue()).toStringUtf8();
+ ClpColumnHandle targetHandle = new ClpColumnHandle(jsonPath, call.getType());
+
+ // Check if a variable with the same ClpColumnHandle already exists
+ VariableReferenceExpression existingVar = globalColumnVarMap.get(targetHandle);
+ if (existingVar != null) {
+ return existingVar;
+ }
+
+ VariableReferenceExpression newVar = variableAllocator.newVariable(
+ expression.getSourceLocation(),
+ encodeJsonPath(jsonPath),
+ call.getType());
+ globalColumnVarMap.put(targetHandle, newVar);
+ return newVar;
+ }
+
+ // Recurse into arguments
+ List rewrittenArgs = call.getArguments().stream()
+ .map(arg -> rewriteClpUdfs(arg, functionManager, variableAllocator))
+ .collect(toImmutableList());
+
+ return new CallExpression(call.getDisplayName(), call.getFunctionHandle(), call.getType(), rewrittenArgs);
+ }
+
+ // Handle special forms (e.g., AND, OR, etc.)
+ if (expression instanceof SpecialFormExpression) {
+ SpecialFormExpression special = (SpecialFormExpression) expression;
+
+ List rewrittenArgs = special.getArguments().stream()
+ .map(arg -> rewriteClpUdfs(arg, functionManager, variableAllocator))
+ .collect(toImmutableList());
+
+ return new SpecialFormExpression(special.getSourceLocation(), special.getForm(), special.getType(), rewrittenArgs);
+ }
+
+ return expression;
+ }
+
+ /**
+ * Recursively rewrites the subtree of a plan node to include any new variables produced by
+ * CLP UDF rewrites.
+ *
+ * @param node the plan node to rewrite
+ * @return the rewritten plan node
+ */
+ private PlanNode rewritePlanSubtree(PlanNode node)
+ {
+ if (node instanceof TableScanNode) {
+ return buildNewTableScanNode((TableScanNode) node);
+ }
+ else if (node instanceof FilterNode) {
+ return buildNewFilterNode((FilterNode) node);
+ }
+
+ List rewrittenChildren = node.getSources().stream()
+ .map(source -> rewritePlanSubtree(source))
+ .collect(toImmutableList());
+
+ return node.replaceChildren(rewrittenChildren);
+ }
+
+ /**
+ * Encodes a JSON path into a valid variable name by replacing uppercase letters with
+ * "_ux", dots with "_dot_", and underscores with "_und_".
+ *
+ * This is only used internally to ensure that the variable names generated from JSON paths
+ * are valid and do not conflict with other variable names in the expression.
+ *
+ * @param jsonPath the JSON path to encode
+ * @return the encoded variable name
+ */
+ private String encodeJsonPath(String jsonPath)
+ {
+ StringBuilder sb = new StringBuilder();
+ for (char c : jsonPath.toCharArray()) {
+ if (Character.isUpperCase(c)) {
+ sb.append("_ux").append(Character.toLowerCase(c));
+ }
+ else if (c == '.') {
+ sb.append("_dot_");
+ }
+ else if (c == '_') {
+ sb.append("_und_");
+ }
+ else {
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Builds a new {@link TableScanNode} that includes additional
+ * {@link VariableReferenceExpression}s and {@link ColumnHandle}s for rewritten CLP UDFs.
+ *
+ * @param node the original table scan node
+ * @return the updated table scan node
+ */
+ private TableScanNode buildNewTableScanNode(TableScanNode node)
+ {
+ Set outputVars = new LinkedHashSet<>(node.getOutputVariables());
+ Map newAssignments = new HashMap<>(node.getAssignments());
+
+ // Add any missing variables for known handles
+ globalColumnVarMap.forEach((handle, var) -> {
+ outputVars.add(var);
+ newAssignments.put(var, handle);
+ });
+
+ return new TableScanNode(
+ node.getSourceLocation(),
+ idAllocator.getNextId(),
+ node.getTable(),
+ new ArrayList<>(outputVars),
+ newAssignments,
+ node.getTableConstraints(),
+ node.getCurrentConstraint(),
+ node.getEnforcedConstraint(),
+ node.getCteMaterializationInfo());
+ }
+
+ /**
+ * Builds a new {@link FilterNode} with its predicate rewritten to replace CLP UDF calls.
+ *
+ * @param node the original filter node
+ * @return the updated filter node
+ */
+ private FilterNode buildNewFilterNode(FilterNode node)
+ {
+ RowExpression newPredicate = rewriteClpUdfs(node.getPredicate(), functionManager, variableAllocator);
+ PlanNode newSource = rewritePlanSubtree(node.getSource());
+ return new FilterNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, newPredicate);
+ }
+ }
+}
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 91fc3c780d941..d1d0ee6964c8e 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
@@ -226,6 +226,11 @@ static final class DbHandle
{
this.dbPath = dbPath;
}
+
+ public String getDbPath()
+ {
+ return dbPath;
+ }
}
static final class ArchivesTableRow
diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpFilterToKql.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpFilterToKql.java
index 9e9e9db13a665..979c9cbfbcf69 100644
--- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpFilterToKql.java
+++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpFilterToKql.java
@@ -13,6 +13,7 @@
*/
package com.facebook.presto.plugin.clp;
+import com.facebook.presto.plugin.clp.optimization.ClpFilterToKqlConverter;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.relation.RowExpression;
import com.facebook.presto.spi.relation.VariableReferenceExpression;
@@ -289,7 +290,10 @@ private void testPushDown(SessionHolder sessionHolder, String sql, String expect
}
}
- private ClpExpression tryPushDown(String sqlExpression, SessionHolder sessionHolder, Set metadataFilterColumns)
+ private ClpExpression tryPushDown(
+ String sqlExpression,
+ SessionHolder sessionHolder,
+ Set metadataFilterColumns)
{
RowExpression pushDownExpression = getRowExpression(sqlExpression, sessionHolder);
Map assignments = new HashMap<>(variableToColumnHandleMap);
@@ -297,11 +301,16 @@ private ClpExpression tryPushDown(String sqlExpression, SessionHolder sessionHol
new ClpFilterToKqlConverter(
standardFunctionResolution,
functionAndTypeManager,
+ assignments,
metadataFilterColumns),
- assignments);
+ null);
}
- private void testFilter(ClpExpression clpExpression, String expectedKqlExpression, String expectedRemainingExpression, SessionHolder sessionHolder)
+ private void testFilter(
+ ClpExpression clpExpression,
+ String expectedKqlExpression,
+ String expectedRemainingExpression,
+ SessionHolder sessionHolder)
{
Optional kqlExpression = clpExpression.getPushDownExpression();
Optional remainingExpression = clpExpression.getRemainingExpression();
diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpQueryBase.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpQueryBase.java
index 87cc35d6d2b4e..ee259d67e17c9 100644
--- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpQueryBase.java
+++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpQueryBase.java
@@ -22,6 +22,7 @@
import com.facebook.presto.metadata.CatalogManager;
import com.facebook.presto.metadata.ColumnPropertyManager;
import com.facebook.presto.metadata.FunctionAndTypeManager;
+import com.facebook.presto.metadata.FunctionExtractor;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.metadata.MetadataManager;
import com.facebook.presto.metadata.SchemaPropertyManager;
@@ -64,6 +65,9 @@
public class TestClpQueryBase
{
protected static final FunctionAndTypeManager functionAndTypeManager = createTestFunctionAndTypeManager();
+ static {
+ functionAndTypeManager.registerBuiltInFunctions(FunctionExtractor.extractFunctions(ClpFunctions.class));
+ }
protected static final StandardFunctionResolution standardFunctionResolution = new FunctionResolution(functionAndTypeManager.getFunctionAndTypeResolver());
protected static final Metadata metadata = new MetadataManager(
functionAndTypeManager,
@@ -79,10 +83,10 @@ public class TestClpQueryBase
protected static final ClpColumnHandle city = new ClpColumnHandle(
"city",
RowType.from(ImmutableList.of(
+ RowType.field("Name", VARCHAR),
RowType.field("Region", RowType.from(ImmutableList.of(
RowType.field("Id", BIGINT),
- RowType.field("Name", VARCHAR)))),
- RowType.field("Name", VARCHAR))));
+ RowType.field("Name", VARCHAR)))))));
protected static final ClpColumnHandle fare = new ClpColumnHandle("fare", DOUBLE);
protected static final ClpColumnHandle isHoliday = new ClpColumnHandle("isHoliday", BOOLEAN);
protected static final Map variableToColumnHandleMap =
@@ -116,11 +120,6 @@ protected RowExpression getRowExpression(String sqlExpression, SessionHolder ses
return toRowExpression(expression(sqlExpression), typeProvider, sessionHolder.getSession());
}
- protected RowExpression getRowExpression(String sqlExpression, TypeProvider typeProvider, SessionHolder sessionHolder)
- {
- return toRowExpression(expression(sqlExpression), typeProvider, sessionHolder.getSession());
- }
-
protected static class SessionHolder
{
private final ConnectorSession connectorSession;
diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpUdfRewriter.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpUdfRewriter.java
new file mode 100644
index 0000000000000..b82866f3d8dd9
--- /dev/null
+++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpUdfRewriter.java
@@ -0,0 +1,324 @@
+/*
+ * 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.Session;
+import com.facebook.presto.common.transaction.TransactionId;
+import com.facebook.presto.common.type.ArrayType;
+import com.facebook.presto.cost.PlanNodeStatsEstimate;
+import com.facebook.presto.cost.StatsAndCosts;
+import com.facebook.presto.cost.StatsProvider;
+import com.facebook.presto.metadata.FunctionAndTypeManager;
+import com.facebook.presto.metadata.Metadata;
+import com.facebook.presto.plugin.clp.optimization.ClpComputePushDown;
+import com.facebook.presto.plugin.clp.optimization.ClpUdfRewriter;
+import com.facebook.presto.plugin.clp.split.filter.ClpMySqlSplitFilterProvider;
+import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider;
+import com.facebook.presto.spi.ColumnHandle;
+import com.facebook.presto.spi.SchemaTableName;
+import com.facebook.presto.spi.VariableAllocator;
+import com.facebook.presto.spi.WarningCollector;
+import com.facebook.presto.spi.plan.PlanNode;
+import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
+import com.facebook.presto.spi.plan.TableScanNode;
+import com.facebook.presto.spi.relation.VariableReferenceExpression;
+import com.facebook.presto.sql.planner.Plan;
+import com.facebook.presto.sql.planner.assertions.MatchResult;
+import com.facebook.presto.sql.planner.assertions.Matcher;
+import com.facebook.presto.sql.planner.assertions.PlanAssert;
+import com.facebook.presto.sql.planner.assertions.PlanMatchPattern;
+import com.facebook.presto.sql.planner.assertions.SymbolAliases;
+import com.facebook.presto.sql.relational.FunctionResolution;
+import com.facebook.presto.sql.tree.SymbolReference;
+import com.facebook.presto.testing.LocalQueryRunner;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.math3.util.Pair;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static com.facebook.presto.common.Utils.checkState;
+import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.common.type.DoubleType.DOUBLE;
+import static com.facebook.presto.common.type.VarcharType.VARCHAR;
+import static com.facebook.presto.metadata.FunctionExtractor.extractFunctions;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.ARCHIVES_STORAGE_DIRECTORY_BASE;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.METADATA_DB_PASSWORD;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.METADATA_DB_TABLE_PREFIX;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.METADATA_DB_URL_TEMPLATE;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.METADATA_DB_USER;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.getDbHandle;
+import static com.facebook.presto.plugin.clp.ClpMetadataDbSetUp.setupMetadata;
+import static com.facebook.presto.plugin.clp.metadata.ClpSchemaTreeNodeType.Boolean;
+import static com.facebook.presto.plugin.clp.metadata.ClpSchemaTreeNodeType.ClpString;
+import static com.facebook.presto.plugin.clp.metadata.ClpSchemaTreeNodeType.Float;
+import static com.facebook.presto.plugin.clp.metadata.ClpSchemaTreeNodeType.Integer;
+import static com.facebook.presto.plugin.clp.metadata.ClpSchemaTreeNodeType.VarString;
+import static com.facebook.presto.sql.planner.assertions.MatchResult.NO_MATCH;
+import static com.facebook.presto.sql.planner.assertions.MatchResult.match;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.anyTree;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.filter;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.node;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.project;
+import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
+import static java.lang.String.format;
+
+@Test(singleThreaded = true)
+public class TestClpUdfRewriter
+ extends TestClpQueryBase
+{
+ private final Session defaultSession = testSessionBuilder()
+ .setCatalog("clp")
+ .setSchema(ClpMetadata.DEFAULT_SCHEMA_NAME)
+ .build();
+
+ private ClpMetadataDbSetUp.DbHandle dbHandle;
+ ClpTableHandle table;
+
+ private LocalQueryRunner localQueryRunner;
+ private FunctionAndTypeManager functionAndTypeManager;
+ private FunctionResolution functionResolution;
+ private ClpSplitFilterProvider splitFilterProvider;
+ private PlanNodeIdAllocator planNodeIdAllocator;
+ private VariableAllocator variableAllocator;
+
+ @BeforeMethod
+ public void setUp()
+ {
+ dbHandle = getDbHandle("metadata_query_testdb");
+ final String tableName = "test";
+ final String tablePath = ARCHIVES_STORAGE_DIRECTORY_BASE + tableName;
+ table = new ClpTableHandle(new SchemaTableName("default", tableName), tablePath);
+
+ setupMetadata(dbHandle,
+ ImmutableMap.of(
+ tableName,
+ ImmutableList.of(
+ 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))));
+
+ 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());
+ planNodeIdAllocator = new PlanNodeIdAllocator();
+ variableAllocator = new VariableAllocator();
+ }
+
+ @AfterMethod
+ public void tearDown()
+ {
+ localQueryRunner.close();
+ ClpMetadataDbSetUp.tearDown(dbHandle);
+ }
+
+ @Test
+ public void testScanFilter()
+ {
+ TransactionId transactionId = localQueryRunner.getTransactionManager().beginTransaction(false);
+ Session session = testSessionBuilder().setCatalog("clp").setSchema("default").setTransactionId(transactionId).build();
+
+ Plan plan = localQueryRunner.createPlan(
+ session,
+ "SELECT * FROM test WHERE CLP_GET_BIGINT('user_id') = 0 AND CLP_GET_DOUBLE('fare') < 50.0 AND CLP_GET_STRING('city') = 'SF' AND " +
+ "CLP_GET_BOOL('isHoliday') = true AND cardinality(CLP_GET_STRING_ARRAY('tags')) > 0 AND LOWER(city.Name) = 'beijing'",
+ WarningCollector.NOOP);
+ ClpUdfRewriter udfRewriter = new ClpUdfRewriter(functionAndTypeManager);
+ PlanNode optimizedPlan = udfRewriter.optimize(plan.getRoot(), session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+ ClpComputePushDown optimizer = new ClpComputePushDown(functionAndTypeManager, functionResolution, splitFilterProvider);
+ optimizedPlan = optimizer.optimize(optimizedPlan, session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+
+ PlanAssert.assertPlan(
+ session,
+ localQueryRunner.getMetadata(),
+ (node, sourceStats, lookup, s, types) -> PlanNodeStatsEstimate.unknown(),
+ new Plan(optimizedPlan, plan.getTypes(), StatsAndCosts.empty()),
+ anyTree(
+ filter(
+ expression("lower(city.Name) = 'beijing' AND cardinality(tags) > 0"),
+ ClpTableScanMatcher.clpTableScanPattern(
+ new ClpTableLayoutHandle(
+ table,
+ Optional.of(
+ "(((user_id: 0 AND fare < 50.0) AND (city: \"SF\" AND isHoliday: true)))"),
+ Optional.empty()),
+ ImmutableSet.of(
+ city,
+ fare,
+ isHoliday,
+ new ClpColumnHandle("user_id", BIGINT),
+ new ClpColumnHandle("city", VARCHAR),
+ new ClpColumnHandle("tags", new ArrayType(VARCHAR)))))));
+ }
+
+ @Test
+ public void testScanProject()
+ {
+ TransactionId transactionId = localQueryRunner.getTransactionManager().beginTransaction(false);
+ Session session = testSessionBuilder().setCatalog("clp").setSchema("default").setTransactionId(transactionId).build();
+
+ Plan plan = localQueryRunner.createPlan(
+ session,
+ "SELECT CLP_GET_BIGINT('user_id'), CLP_GET_DOUBLE('fare'), CLP_GET_STRING('user'), " +
+ "CLP_GET_BOOL('isHoliday'), CLP_GET_STRING_ARRAY('tags'), city.Name FROM test",
+ WarningCollector.NOOP);
+ ClpUdfRewriter udfRewriter = new ClpUdfRewriter(functionAndTypeManager);
+ PlanNode optimizedPlan = udfRewriter.optimize(plan.getRoot(), session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+ ClpComputePushDown optimizer = new ClpComputePushDown(functionAndTypeManager, functionResolution, splitFilterProvider);
+ optimizedPlan = optimizer.optimize(optimizedPlan, session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+
+ PlanAssert.assertPlan(
+ session,
+ localQueryRunner.getMetadata(),
+ (node, sourceStats, lookup, s, types) -> PlanNodeStatsEstimate.unknown(),
+ new Plan(optimizedPlan, plan.getTypes(), StatsAndCosts.empty()),
+ anyTree(
+ project(
+ ImmutableMap.of(
+ "clp_get_bigint",
+ PlanMatchPattern.expression("user_und_id"),
+ "clp_get_double",
+ PlanMatchPattern.expression("fare"),
+ "clp_get_string",
+ PlanMatchPattern.expression("user"),
+ "clp_get_bool",
+ PlanMatchPattern.expression("is_uxholiday"),
+ "clp_get_string_array",
+ PlanMatchPattern.expression("tags"),
+ "expr",
+ PlanMatchPattern.expression("city.Name")),
+ ClpTableScanMatcher.clpTableScanPattern(
+ new ClpTableLayoutHandle(
+ table,
+ Optional.empty(),
+ Optional.empty()),
+ ImmutableSet.of(
+ new ClpColumnHandle("user_id", BIGINT),
+ new ClpColumnHandle("fare", DOUBLE),
+ new ClpColumnHandle("user", VARCHAR),
+ isHoliday,
+ new ClpColumnHandle("tags", new ArrayType(VARCHAR)),
+ city)))));
+ }
+
+ @Test
+ public void testScanProjectFilter()
+ {
+ TransactionId transactionId = localQueryRunner.getTransactionManager().beginTransaction(false);
+ Session session = testSessionBuilder().setCatalog("clp").setSchema("default").setTransactionId(transactionId).build();
+
+ Plan plan = localQueryRunner.createPlan(
+ session,
+ "SELECT LOWER(city.Name), LOWER(CLP_GET_STRING('city.Name')) from test WHERE CLP_GET_BIGINT('user_id') = 0 AND LOWER(city.Name) = 'beijing'",
+ WarningCollector.NOOP);
+ ClpUdfRewriter udfRewriter = new ClpUdfRewriter(functionAndTypeManager);
+ PlanNode optimizedPlan = udfRewriter.optimize(plan.getRoot(), session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+ ClpComputePushDown optimizer = new ClpComputePushDown(functionAndTypeManager, functionResolution, splitFilterProvider);
+ optimizedPlan = optimizer.optimize(optimizedPlan, session.toConnectorSession(), variableAllocator, planNodeIdAllocator);
+
+ PlanAssert.assertPlan(
+ session,
+ localQueryRunner.getMetadata(),
+ (node, sourceStats, lookup, s, types) -> PlanNodeStatsEstimate.unknown(),
+ new Plan(optimizedPlan, plan.getTypes(), StatsAndCosts.empty()),
+ anyTree(
+ project(
+ ImmutableMap.of(
+ "lower",
+ PlanMatchPattern.expression("lower(city.Name)"),
+ "lower_0",
+ PlanMatchPattern.expression("lower(city_dot__uxname)")),
+ filter(
+ expression("lower(city.Name) = 'beijing'"),
+ ClpTableScanMatcher.clpTableScanPattern(
+ new ClpTableLayoutHandle(table, Optional.of("(user_id: 0)"), Optional.empty()),
+ ImmutableSet.of(
+ new ClpColumnHandle("city.Name", VARCHAR),
+ new ClpColumnHandle("user_id", BIGINT),
+ city))))));
+ }
+
+ 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-native-execution/velox b/presto-native-execution/velox
index 7c55762bc37ed..ab8f44627de6a 160000
--- a/presto-native-execution/velox
+++ b/presto-native-execution/velox
@@ -1 +1 @@
-Subproject commit 7c55762bc37edc3f93cb8f5f1ff625e06cc10deb
+Subproject commit ab8f44627de6ac2fe46fb5ca8ed3af09ee477b42