You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2022/08/05 02:31:49 UTC

Change in asterixdb[cheshire-cat]: WIP: dynamic prefixes

From Ali Alsuliman <al...@gmail.com>:

Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17083 )


Change subject: WIP: dynamic prefixes
......................................................................

WIP: dynamic prefixes

Change-Id: I253680032e5b6444f3025ed874fe776034401aca
---
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
A asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/FindExternalDatasetPrefixRule.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStreamFactory.java
3 files changed, 191 insertions(+), 0 deletions(-)



  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/83/17083/1

diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
index fc52f89..f21519a 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/RuleCollections.java
@@ -46,6 +46,7 @@
 import org.apache.asterix.optimizer.rules.ExtractWindowExpressionsRule;
 import org.apache.asterix.optimizer.rules.FeedScanCollectionToUnnest;
 import org.apache.asterix.optimizer.rules.FindDataSourcesRule;
+import org.apache.asterix.optimizer.rules.FindExternalDatasetPrefixRule;
 import org.apache.asterix.optimizer.rules.FixReplicateOperatorOutputsRule;
 import org.apache.asterix.optimizer.rules.FullTextContainsParameterCheckRule;
 import org.apache.asterix.optimizer.rules.FuzzyEqRule;
@@ -212,6 +213,7 @@
         normalization.add(new RemoveRedundantSelectRule());
         normalization.add(new CancelUnnestSingletonListRule());
         normalization.add(new UnnestToDataScanRule());
+        normalization.add(new FindExternalDatasetPrefixRule());
         normalization.add(new MetaFunctionToMetaVariableRule());
         normalization.add(new FuzzyEqRule());
         normalization.add(new SimilarityCheckRule());
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/FindExternalDatasetPrefixRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/FindExternalDatasetPrefixRule.java
new file mode 100644
index 0000000..bf1285d
--- /dev/null
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/FindExternalDatasetPrefixRule.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.asterix.optimizer.rules;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.metadata.declared.DataSource;
+import org.apache.asterix.metadata.declared.DatasetDataSource;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.entities.ExternalDatasetDetails;
+import org.apache.asterix.om.functions.BuiltinFunctions;
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IAlgebricksConstantValue;
+import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
+import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.SelectOperator;
+import org.apache.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
+import org.apache.hyracks.util.JSONUtil;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+public class FindExternalDatasetPrefixRule implements IAlgebraicRewriteRule {
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    @Override
+    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        return false;
+    }
+
+    @Override
+    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+            throws AlgebricksException {
+        AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
+        if (op.getOperatorTag() != LogicalOperatorTag.SELECT) {
+            return false;
+        }
+        if (context.checkIfInDontApplySet(this, op)) {
+            return false;
+        }
+        context.addToDontApplySet(this, op);
+        SelectOperator select = (SelectOperator) op;
+        List<AbstractFunctionCallExpression> eqFunctions = new ArrayList<>();
+        findPrefixes(select.getCondition().getValue(), eqFunctions);
+        if (eqFunctions.isEmpty()) {
+            return false;
+        }
+        List<DataSourceScanOperator> externalScanOps = new ArrayList<>();
+        findExternalDataSources(op, externalScanOps);
+        addExternalPrefixInfo(externalScanOps, eqFunctions);
+        return false;
+    }
+
+    private void addExternalPrefixInfo(List<DataSourceScanOperator> externalScanOps,
+            List<AbstractFunctionCallExpression> eqFunctions) throws AlgebricksException {
+        for (int i = 0; i < externalScanOps.size(); i++) {
+            DataSourceScanOperator externalScanOp = externalScanOps.get(i);
+            LogicalVariable scanVar = ((DatasetDataSource) externalScanOp.getDataSource())
+                    .getDataRecordVariable(externalScanOp.getScanVariables());
+            ObjectNode prefixes = JSONUtil.createObject();
+            for (int k = 0; k < eqFunctions.size(); k++) {
+                List<Mutable<ILogicalExpression>> eqArgs = eqFunctions.get(k).getArguments();
+                ILogicalExpression arg0 = eqArgs.get(0).getValue();
+                ILogicalExpression arg1 = eqArgs.get(1).getValue();
+                AbstractFunctionCallExpression fieldAccess;
+                ConstantExpression constant;
+                if (arg0.getExpressionTag() == LogicalExpressionTag.CONSTANT) {
+                    fieldAccess = (AbstractFunctionCallExpression) arg1;
+                    constant = (ConstantExpression) arg0;
+                } else {
+                    fieldAccess = (AbstractFunctionCallExpression) arg0;
+                    constant = (ConstantExpression) arg1;
+                }
+                IAlgebricksConstantValue value = constant.getValue();
+                ILogicalExpression expr = fieldAccess.getArguments().get(0).getValue();
+                VariableReferenceExpression varRefExpr = (VariableReferenceExpression) expr;
+                if (scanVar.equals(varRefExpr.getVariableReference())) {
+                    ILogicalExpression field = fieldAccess.getArguments().get(1).getValue();
+                    prefixes.put(stripQuotes(field.toString()), stripQuotes(value.toString()));
+                }
+            }
+            Dataset dataset = ((DatasetDataSource) externalScanOp.getDataSource()).getDataset();
+            ExternalDatasetDetails datasetDetails = (ExternalDatasetDetails) dataset.getDatasetDetails();
+            if (!prefixes.isEmpty()) {
+                Map<String, String> properties = datasetDetails.getProperties();
+                if (properties.containsKey("prefixes")) {
+                    try {
+                        JsonNode jsonNode = mapper.readTree(properties.get("prefixes"));
+                        prefixes.setAll((ObjectNode) jsonNode);
+                    } catch (JsonProcessingException e) {
+                        throw new AlgebricksException(e);
+                    }
+                }
+                properties.put("prefixes", prefixes.toString());
+            }
+        }
+    }
+
+    private void findPrefixes(ILogicalExpression expr, List<AbstractFunctionCallExpression> eqFunctions) {
+        if (expr.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
+            return;
+        }
+        AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) expr;
+        List<Mutable<ILogicalExpression>> arguments = funcExpr.getArguments();
+        int numArgs = arguments.size();
+        for (int i = 0; i < numArgs; i++) {
+            findPrefixes(arguments.get(i).getValue(), eqFunctions);
+        }
+        FunctionIdentifier functionIdentifier = funcExpr.getFunctionIdentifier();
+        if (functionIdentifier.equals(BuiltinFunctions.EQ)) {
+            if (eligibleEQ(funcExpr)) {
+                eqFunctions.add(funcExpr);
+            }
+        }
+    }
+
+    private boolean eligibleEQ(AbstractFunctionCallExpression eq) {
+        List<Mutable<ILogicalExpression>> arguments = eq.getArguments();
+        if (arguments.size() != 2) {
+            return false;
+        }
+        Mutable<ILogicalExpression> arg0 = arguments.get(0);
+        Mutable<ILogicalExpression> arg1 = arguments.get(1);
+        return (arg0.getValue().getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL
+                && arg1.getValue().getExpressionTag() == LogicalExpressionTag.CONSTANT)
+                || (arg1.getValue().getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL
+                        && arg0.getValue().getExpressionTag() == LogicalExpressionTag.CONSTANT);
+    }
+
+    private void findExternalDataSources(ILogicalOperator startOp, List<DataSourceScanOperator> dataScanOps) {
+        List<Mutable<ILogicalOperator>> opInputs = startOp.getInputs();
+        for (int i = 0; i < opInputs.size(); i++) {
+            ILogicalOperator child = opInputs.get(i).getValue();
+            findExternalDataSources(child, dataScanOps);
+        }
+        if (startOp.getOperatorTag() == LogicalOperatorTag.DATASOURCESCAN) {
+            DataSourceScanOperator scanOp = (DataSourceScanOperator) startOp;
+            if (((DataSource) scanOp.getDataSource()).getDatasourceType() == DataSource.Type.EXTERNAL_DATASET) {
+                dataScanOps.add(scanOp);
+            }
+        }
+    }
+
+    private static String stripQuotes(String input) {
+        if (input.length() >= 2 && input.startsWith("\"") && input.endsWith("\"")) {
+            return input.substring(1, input.length() - 1);
+        } else {
+            return input;
+        }
+    }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStreamFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStreamFactory.java
index 8197524..ae8b6ad 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStreamFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/aws/AwsS3InputStreamFactory.java
@@ -72,6 +72,11 @@
         String container = configuration.get(ExternalDataConstants.CONTAINER_NAME_FIELD_NAME);
         S3Client s3Client = ExternalDataUtils.AwsS3.buildAwsS3Client(configuration);
 
+        if (configuration.containsKey("prefixes")) {
+            // handle {"field1": 7, "field2": 5}
+            String jsonPrefixes = configuration.get("prefixes");
+        }
+
         try {
             filesOnly = listS3Objects(s3Client, container, includeExcludeMatcher);
         } catch (S3Exception ex) {

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17083
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-Project: asterixdb
Gerrit-Branch: cheshire-cat
Gerrit-Change-Id: I253680032e5b6444f3025ed874fe776034401aca
Gerrit-Change-Number: 17083
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-MessageType: newchange