[CALCITE-1598] Pig adapter (Eli Levine) The adapter will probably be mostly used offline, but PigSchema and PigToEnumerableConverter exist in order to make the Pig adapter behave like an actual adapter.
Close apache/calcite#365 Project: http://git-wip-us.apache.org/repos/asf/calcite/repo Commit: http://git-wip-us.apache.org/repos/asf/calcite/commit/fdbb81cf Tree: http://git-wip-us.apache.org/repos/asf/calcite/tree/fdbb81cf Diff: http://git-wip-us.apache.org/repos/asf/calcite/diff/fdbb81cf Branch: refs/heads/master Commit: fdbb81cf23016887ad4df0423080dbe8ebd68526 Parents: 0372d23 Author: Eli Levine <[email protected]> Authored: Thu Jan 26 15:40:40 2017 -0800 Committer: Julian Hyde <[email protected]> Committed: Tue Feb 28 17:14:09 2017 -0800 ---------------------------------------------------------------------- .../org/apache/calcite/util/BuiltInMethod.java | 1 + pig/pom.xml | 149 ++++++++++ .../calcite/adapter/pig/PigAggFunction.java | 51 ++++ .../calcite/adapter/pig/PigAggregate.java | 206 ++++++++++++++ .../apache/calcite/adapter/pig/PigDataType.java | 68 +++++ .../apache/calcite/adapter/pig/PigFilter.java | 142 ++++++++++ .../org/apache/calcite/adapter/pig/PigJoin.java | 114 ++++++++ .../apache/calcite/adapter/pig/PigProject.java | 58 ++++ .../org/apache/calcite/adapter/pig/PigRel.java | 91 +++++++ .../calcite/adapter/pig/PigRelFactories.java | 118 ++++++++ .../apache/calcite/adapter/pig/PigRules.java | 144 ++++++++++ .../apache/calcite/adapter/pig/PigSchema.java | 41 +++ .../calcite/adapter/pig/PigSchemaFactory.java | 45 ++++ .../apache/calcite/adapter/pig/PigTable.java | 75 ++++++ .../calcite/adapter/pig/PigTableFactory.java | 55 ++++ .../calcite/adapter/pig/PigTableScan.java | 92 +++++++ .../adapter/pig/PigToEnumerableConverter.java | 84 ++++++ .../pig/PigToEnumerableConverterRule.java | 43 +++ .../calcite/adapter/pig/package-info.java | 27 ++ .../apache/calcite/test/AbstractPigTest.java | 36 +++ .../org/apache/calcite/test/PigAdapterTest.java | 190 +++++++++++++ .../calcite/test/PigRelBuilderStyleTest.java | 269 +++++++++++++++++++ pig/src/test/resources/data.txt | 3 + pig/src/test/resources/data2.txt | 2 + pig/src/test/resources/model.json | 47 ++++ pom.xml | 21 +- site/_docs/adapter.md | 5 + site/_docs/pig.md | 90 +++++++ 28 files changed, 2266 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java index 535ea49..458444b 100644 --- a/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java +++ b/core/src/main/java/org/apache/calcite/util/BuiltInMethod.java @@ -174,6 +174,7 @@ public enum BuiltInMethod { SKIP(ExtendedEnumerable.class, "skip", int.class), TAKE(ExtendedEnumerable.class, "take", int.class), SINGLETON_ENUMERABLE(Linq4j.class, "singletonEnumerable", Object.class), + EMPTY_ENUMERABLE(Linq4j.class, "emptyEnumerable"), NULLS_COMPARATOR(Functions.class, "nullsComparator", boolean.class, boolean.class), ARRAY_COMPARER(Functions.class, "arrayComparer"), http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/pom.xml ---------------------------------------------------------------------- diff --git a/pig/pom.xml b/pig/pom.xml new file mode 100644 index 0000000..169cdc7 --- /dev/null +++ b/pig/pom.xml @@ -0,0 +1,149 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- +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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite</artifactId> + <version>1.12.0-SNAPSHOT</version> + </parent> + + <artifactId>calcite-pig</artifactId> + <packaging>jar</packaging> + <version>1.12.0-SNAPSHOT</version> + <name>Calcite Pig</name> + <description>Pig adapter for Calcite</description> + + <properties> + <top.dir>${project.basedir}/..</top.dir> + </properties> + + <dependencies> + <!-- Sorted by groupId, artifactId; calcite dependencies first. Put versions + in dependencyManagement in the root POM, not here. --> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-core</artifactId> + <type>jar</type> + </dependency> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-core</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.calcite</groupId> + <artifactId>calcite-linq4j</artifactId> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-client</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.pig</groupId> + <artifactId>pig</artifactId> + <classifier>h2</classifier> + </dependency> + <dependency> + <groupId>org.apache.pig</groupId> + <artifactId>pigunit</artifactId> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </dependency> + <dependency> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <!-- Sorted by groupId, artifactId. Put versions in + pluginManagement in the root POM, not here. --> + <plugin> + <artifactId>maven-dependency-plugin</artifactId> + <version>${maven-dependency-plugin.version}</version> + <executions> + <execution> + <id>analyze</id> + <goals> + <goal>analyze-only</goal> + </goals> + <configuration> + <failOnWarning>true</failOnWarning> + <!-- ignore "unused but declared" warnings --> + <ignoredUnusedDeclaredDependencies> + <ignoredUnusedDeclaredDependency>org.slf4j:slf4j-api</ignoredUnusedDeclaredDependency> + <ignoredUnusedDeclaredDependency>org.slf4j:slf4j-log4j12</ignoredUnusedDeclaredDependency> + <ignoredUnusedDeclaredDependency>org.apache.hadoop:hadoop-client</ignoredUnusedDeclaredDependency> + </ignoredUnusedDeclaredDependencies> + </configuration> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + <executions> + <execution> + <goals> + <goal>test-jar</goal> + </goals> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-release-plugin</artifactId> + </plugin> + <!-- Parent module has the same plugin and does the work of + generating -sources.jar for each project. But without the + plugin declared here, IDEs don't know the sources are + available. --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-source-plugin</artifactId> + <executions> + <execution> + <id>attach-sources</id> + <phase>verify</phase> + <goals> + <goal>jar-no-fork</goal> + <goal>test-jar-no-fork</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> + +</project> http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggFunction.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggFunction.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggFunction.java new file mode 100644 index 0000000..da88b4e --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggFunction.java @@ -0,0 +1,51 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.sql.SqlKind; + +/** + * Supported Pig aggregate functions and their Calcite counterparts. The enum's + * name() is the same as the function's name in Pig Latin. + */ +public enum PigAggFunction { + + COUNT(SqlKind.COUNT, false), COUNT_STAR(SqlKind.COUNT, true); + + private final SqlKind calciteFunc; + private final boolean star; // as in COUNT(*) + + private PigAggFunction(SqlKind calciteFunc) { + this(calciteFunc, false); + } + + private PigAggFunction(SqlKind calciteFunc, boolean star) { + this.calciteFunc = calciteFunc; + this.star = star; + } + + public static PigAggFunction valueOf(SqlKind calciteFunc, boolean star) { + for (PigAggFunction pigAggFunction : values()) { + if (pigAggFunction.calciteFunc == calciteFunc && pigAggFunction.star == star) { + return pigAggFunction; + } + } + throw new IllegalArgumentException("Pig agg func for " + calciteFunc + " is not supported"); + } +} + +// End PigAggFunction.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggregate.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggregate.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggregate.java new file mode 100644 index 0000000..1f83306 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigAggregate.java @@ -0,0 +1,206 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Aggregate; +import org.apache.calcite.rel.core.AggregateCall; +import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.util.ImmutableBitSet; + +import org.apache.pig.scripting.Pig; + +import com.google.common.base.Joiner; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** Implementation of {@link org.apache.calcite.rel.core.Aggregate} in + * {@link PigRel#CONVENTION Pig calling convention}. */ +public class PigAggregate extends Aggregate implements PigRel { + + public static final String DISTINCT_FIELD_SUFFIX = "_DISTINCT"; + + /** Creates a PigAggregate. */ + public PigAggregate(RelOptCluster cluster, RelTraitSet traits, RelNode child, boolean indicator, + ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) { + super(cluster, traits, child, indicator, groupSet, groupSets, aggCalls); + assert getConvention() == PigRel.CONVENTION; + //assert getConvention() == child.getConvention(); + } + + @Override public Aggregate copy(RelTraitSet traitSet, RelNode input, boolean indicator, + ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) { + return new PigAggregate(input.getCluster(), traitSet, input, indicator, groupSet, groupSets, + aggCalls); + } + + @Override public void implement(Implementor implementor) { + implementor.visitChild(0, getInput()); + implementor.addStatement(getPigAggregateStatement(implementor)); + } + + /** + * Generates a GROUP BY statement, followed by an optional FOREACH statement + * for all aggregate functions used. e.g. + * <pre> + * {@code + * A = GROUP A BY owner; + * A = FOREACH A GENERATE group, SUM(A.pet_num); + * } + * </pre> + */ + private String getPigAggregateStatement(Implementor implementor) { + return getPigGroupBy(implementor) + '\n' + getPigForEachGenerate(implementor); + } + + /** + * Override this method so it looks down the tree to find the table this node + * is acting on. + */ + @Override public RelOptTable getTable() { + return getInput().getTable(); + } + + /** + * Generates the GROUP BY statement, e.g. + * <code>A = GROUP A BY (f1, f2);</code> + */ + private String getPigGroupBy(Implementor implementor) { + final String relAlias = implementor.getPigRelationAlias(this); + final List<RelDataTypeField> allFields = getInput().getRowType().getFieldList(); + final List<Integer> groupedFieldIndexes = groupSet.asList(); + if (groupedFieldIndexes.size() < 1) { + return relAlias + " = GROUP " + relAlias + " ALL;"; + } else { + final List<String> groupedFieldNames = new ArrayList<>(groupedFieldIndexes.size()); + for (int fieldIndex : groupedFieldIndexes) { + groupedFieldNames.add(allFields.get(fieldIndex).getName()); + } + return relAlias + " = GROUP " + relAlias + " BY (" + Joiner.on(", ").join(groupedFieldNames) + + ");"; + } + } + + /** + * Generates a FOREACH statement containing invocation of aggregate functions + * and projection of grouped fields. e.g. + * <code>A = FOREACH A GENERATE group, SUM(A.pet_num);</code> + * @see Pig documentation for special meaning of the "group" field after GROUP + * BY. + */ + private String getPigForEachGenerate(Implementor implementor) { + final String relAlias = implementor.getPigRelationAlias(this); + final String generateCall = getPigGenerateCall(implementor); + final List<String> distinctCalls = getDistinctCalls(implementor); + return relAlias + " = FOREACH " + relAlias + " {\n" + + Joiner.on(";\n").join(distinctCalls) + generateCall + "\n};"; + } + + private String getPigGenerateCall(Implementor implementor) { + final List<Integer> groupedFieldIndexes = groupSet.asList(); + Set<String> groupFields = new HashSet<>(groupedFieldIndexes.size()); + for (int fieldIndex : groupedFieldIndexes) { + final String fieldName = getInputFieldName(fieldIndex); + // Pig appends group field name if grouping by multiple fields + final String groupField = (groupedFieldIndexes.size() == 1 ? "group" : ("group." + fieldName)) + + " AS " + fieldName; + + groupFields.add(groupField); + } + final List<String> pigAggCalls = getPigAggregateCalls(implementor); + List<String> allFields = new ArrayList<>(groupFields.size() + pigAggCalls.size()); + allFields.addAll(groupFields); + allFields.addAll(pigAggCalls); + return " GENERATE " + Joiner.on(", ").join(allFields) + ';'; + } + + private List<String> getPigAggregateCalls(Implementor implementor) { + final String relAlias = implementor.getPigRelationAlias(this); + final List<String> result = new ArrayList<>(aggCalls.size()); + for (AggregateCall ac : aggCalls) { + result.add(getPigAggregateCall(relAlias, ac)); + } + return result; + } + + private String getPigAggregateCall(String relAlias, AggregateCall aggCall) { + final PigAggFunction aggFunc = toPigAggFunc(aggCall); + final String alias = aggCall.getName(); + final String fields = Joiner.on(", ").join(getArgNames(relAlias, aggCall)); + return aggFunc.name() + "(" + fields + ") AS " + alias; + } + + private PigAggFunction toPigAggFunc(AggregateCall aggCall) { + return PigAggFunction.valueOf(aggCall.getAggregation().getKind(), + aggCall.getArgList().size() < 1); + } + + private List<String> getArgNames(String relAlias, AggregateCall aggCall) { + final List<String> result = new ArrayList<>(aggCall.getArgList().size()); + for (int fieldIndex : aggCall.getArgList()) { + result.add(getInputFieldNameForAggCall(relAlias, aggCall, fieldIndex)); + } + return result; + } + + private String getInputFieldNameForAggCall(String relAlias, AggregateCall aggCall, + int fieldIndex) { + final String inputField = getInputFieldName(fieldIndex); + return aggCall.isDistinct() ? (inputField + DISTINCT_FIELD_SUFFIX) + : (relAlias + '.' + inputField); + } + + /** + * A agg function call like <code>COUNT(DISTINCT COL)</code> in Pig is + * achieved via two statements in a FOREACH that follows a GROUP statement: + * + * <blockquote> + * <code> + * TABLE = GROUP TABLE ALL;<br> + * TABLE = FOREACH TABLE {<br> + * <b>COL.DISTINCT = DISTINCT COL;<br> + * GENERATE COUNT(COL.DISTINCT) AS C;</b><br> + * }</code> + * </blockquote> + */ + private List<String> getDistinctCalls(Implementor implementor) { + final String relAlias = implementor.getPigRelationAlias(this); + final List<String> result = new ArrayList<>(); + for (AggregateCall aggCall : aggCalls) { + if (aggCall.isDistinct()) { + for (int fieldIndex : aggCall.getArgList()) { + String fieldName = getInputFieldName(fieldIndex); + result.add(" " + fieldName + DISTINCT_FIELD_SUFFIX + " = DISTINCT " + relAlias + '.' + + fieldName + ";\n"); + } + } + } + return result; + } + + private String getInputFieldName(int fieldIndex) { + return getInput().getRowType().getFieldList().get(fieldIndex).getName(); + } +} + +// End PigAggregate.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigDataType.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigDataType.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigDataType.java new file mode 100644 index 0000000..a552fb8 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigDataType.java @@ -0,0 +1,68 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.sql.type.SqlTypeName; + +import org.apache.pig.data.DataType; + +import static org.apache.calcite.sql.type.SqlTypeName.VARCHAR; + +/** + * Supported Pig data types and their Calcite counterparts. + */ +public enum PigDataType { + + CHARARRAY(DataType.CHARARRAY, VARCHAR); + + private byte pigType; // Pig defines types using bytes + private SqlTypeName sqlType; + + private PigDataType(byte pigType, SqlTypeName sqlType) { + this.pigType = pigType; + this.sqlType = sqlType; + } + + public byte getPigType() { + return pigType; + } + + public SqlTypeName getSqlType() { + return sqlType; + } + + public static PigDataType valueOf(byte pigType) { + for (PigDataType pigDataType : values()) { + if (pigDataType.pigType == pigType) { + return pigDataType; + } + } + throw new IllegalArgumentException( + "Pig data type " + DataType.findTypeName(pigType) + " is not supported"); + } + + public static PigDataType valueOf(SqlTypeName sqlType) { + for (PigDataType pigDataType : values()) { + if (pigDataType.sqlType == sqlType) { + return pigDataType; + } + } + throw new IllegalArgumentException("SQL data type " + sqlType + " is not supported"); + } +} + +// End PigDataType.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigFilter.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigFilter.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigFilter.java new file mode 100644 index 0000000..63cc1ea --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigFilter.java @@ -0,0 +1,142 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Filter; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexLiteral; +import org.apache.calcite.rex.RexNode; + +import com.google.common.base.Joiner; +import com.google.common.base.Preconditions; + +import java.util.ArrayList; +import java.util.List; + +import static org.apache.calcite.sql.SqlKind.INPUT_REF; +import static org.apache.calcite.sql.SqlKind.LITERAL; + +/** Implementation of {@link org.apache.calcite.rel.core.Filter} in + * {@link PigRel#CONVENTION Pig calling convention}. */ +public class PigFilter extends Filter implements PigRel { + + /** Creates a PigFilter. */ + public PigFilter(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, RexNode condition) { + super(cluster, traitSet, input, condition); + assert getConvention() == PigRel.CONVENTION; + } + + @Override public Filter copy(RelTraitSet traitSet, RelNode input, RexNode condition) { + return new PigFilter(getCluster(), traitSet, input, condition); + } + + @Override public void implement(Implementor implementor) { + implementor.visitChild(0, getInput()); + implementor.addStatement(getPigFilterStatement(implementor)); + } + + /** + * Override this method so it looks down the tree to find the table this node + * is acting on. + */ + @Override public RelOptTable getTable() { + return getInput().getTable(); + } + + /** + * Generates Pig Latin filtering statements, for example + * + * <blockquote> + * <pre>table = FILTER table BY score > 2.0;</pre> + * </blockquote> + */ + private String getPigFilterStatement(Implementor implementor) { + Preconditions.checkState(containsOnlyConjunctions(condition)); + String relationAlias = implementor.getPigRelationAlias(this); + List<String> filterConditionsConjunction = new ArrayList<>(); + for (RexNode node : RelOptUtil.conjunctions(condition)) { + filterConditionsConjunction.add(getSingleFilterCondition(implementor, node)); + } + String allFilterConditions = Joiner.on(" AND ").join(filterConditionsConjunction); + return relationAlias + " = FILTER " + relationAlias + " BY " + allFilterConditions + ';'; + } + + private String getSingleFilterCondition(Implementor implementor, RexNode node) { + switch (node.getKind()) { + case EQUALS: + return getSingleFilterCondition(implementor, "==", (RexCall) node); + case LESS_THAN: + return getSingleFilterCondition(implementor, "<", (RexCall) node); + case LESS_THAN_OR_EQUAL: + return getSingleFilterCondition(implementor, "<=", (RexCall) node); + case GREATER_THAN: + return getSingleFilterCondition(implementor, ">", (RexCall) node); + case GREATER_THAN_OR_EQUAL: + return getSingleFilterCondition(implementor, ">=", (RexCall) node); + default: + throw new IllegalArgumentException("Cannot translate node " + node); + } + } + + private String getSingleFilterCondition(Implementor implementor, String op, RexCall call) { + final String fieldName; + final String literal; + final RexNode left = call.operands.get(0); + final RexNode right = call.operands.get(1); + if (left.getKind() == LITERAL) { + if (right.getKind() != INPUT_REF) { + throw new IllegalArgumentException( + "Expected a RexCall with a single field and single literal"); + } else { + fieldName = implementor.getFieldName(this, ((RexInputRef) right).getIndex()); + literal = getLiteralAsString((RexLiteral) left); + } + } else if (right.getKind() == LITERAL) { + if (left.getKind() != INPUT_REF) { + throw new IllegalArgumentException( + "Expected a RexCall with a single field and single literal"); + } else { + fieldName = implementor.getFieldName(this, ((RexInputRef) left).getIndex()); + literal = getLiteralAsString((RexLiteral) right); + } + } else { + throw new IllegalArgumentException( + "Expected a RexCall with a single field and single literal"); + } + + return '(' + fieldName + ' ' + op + ' ' + literal + ')'; + } + + private boolean containsOnlyConjunctions(RexNode condition) { + return RelOptUtil.disjunctions(condition).size() == 1; + } + + /** + * TODO: do proper literal to string conversion + escaping + */ + private String getLiteralAsString(RexLiteral literal) { + return '\'' + RexLiteral.stringValue(literal) + '\''; + } +} + +// End PigFilter.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigJoin.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigJoin.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigJoin.java new file mode 100644 index 0000000..8880bb1 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigJoin.java @@ -0,0 +1,114 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelOptUtil; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.CorrelationId; +import org.apache.calcite.rel.core.Join; +import org.apache.calcite.rel.core.JoinRelType; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.SqlKind; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; + +/** Implementation of {@link org.apache.calcite.rel.core.Join} in + * {@link PigRel#CONVENTION Pig calling convention}. */ +public class PigJoin extends Join implements PigRel { + + /** Creates a PigJoin. */ + public PigJoin(RelOptCluster cluster, RelTraitSet traitSet, RelNode left, RelNode right, + RexNode condition, JoinRelType joinType) { + super(cluster, traitSet, left, right, condition, new HashSet<CorrelationId>(0), joinType); + assert getConvention() == PigRel.CONVENTION; + } + + @Override public Join copy(RelTraitSet traitSet, RexNode conditionExpr, RelNode left, + RelNode right, JoinRelType joinType, boolean semiJoinDone) { + return new PigJoin(getCluster(), traitSet, left, right, conditionExpr, joinType); + } + + @Override public void implement(Implementor implementor) { + implementor.visitChild(0, getLeft()); + implementor.visitChild(0, getRight()); + implementor.addStatement(getPigJoinStatement(implementor)); + } + + /** + * The Pig alias of the joined relation will have the same name as one from + * the left side of the join. + */ + @Override public RelOptTable getTable() { + return getLeft().getTable(); + } + + /** + * Constructs a Pig JOIN statement in the form of + * <pre> + * {@code + * A = JOIN A BY f1 LEFT OUTER, B BY f2; + * } + * </pre> + * Only supports simple equi-joins with single column on both sides of + * <code>=</code>. + */ + private String getPigJoinStatement(Implementor implementor) { + if (!getCondition().isA(SqlKind.EQUALS)) { + throw new IllegalArgumentException("Only equi-join are supported"); + } + List<RexNode> operands = ((RexCall) getCondition()).getOperands(); + if (operands.size() != 2) { + throw new IllegalArgumentException("Only equi-join are supported"); + } + List<Integer> leftKeys = new ArrayList<>(1); + List<Integer> rightKeys = new ArrayList<>(1); + List<Boolean> filterNulls = new ArrayList<>(1); + RelOptUtil.splitJoinCondition(getLeft(), getRight(), getCondition(), leftKeys, rightKeys, + filterNulls); + + String leftRelAlias = implementor.getPigRelationAlias((PigRel) getLeft()); + String rightRelAlias = implementor.getPigRelationAlias((PigRel) getRight()); + String leftJoinFieldName = implementor.getFieldName((PigRel) getLeft(), leftKeys.get(0)); + String rightJoinFieldName = implementor.getFieldName((PigRel) getRight(), rightKeys.get(0)); + + return implementor.getPigRelationAlias((PigRel) getLeft()) + " = JOIN " + leftRelAlias + " BY " + + leftJoinFieldName + ' ' + getPigJoinType() + ", " + rightRelAlias + " BY " + + rightJoinFieldName + ';'; + } + + /** + * Get a string representation of the type of join for use in a Pig script. + * Pig does not have an explicit "inner" marker, so return an empty string in + * this case. + */ + private String getPigJoinType() { + switch (getJoinType()) { + case INNER: + return ""; + default: + return getJoinType().name(); + } + } +} + +// End PigJoin.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigProject.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigProject.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigProject.java new file mode 100644 index 0000000..fe44e47 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigProject.java @@ -0,0 +1,58 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.Project; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexNode; + +import java.util.List; + +/** Implementation of {@link org.apache.calcite.rel.core.Project} in + * {@link PigRel#CONVENTION Pig calling convention}. */ +public class PigProject extends Project implements PigRel { + + /** Creates a PigProject. */ + public PigProject(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, + List<? extends RexNode> projects, RelDataType rowType) { + super(cluster, traitSet, input, projects, rowType); + assert getConvention() == PigRel.CONVENTION; + } + + @Override public Project copy(RelTraitSet traitSet, RelNode input, List<RexNode> projects, + RelDataType rowType) { + return new PigProject(input.getCluster(), traitSet, input, projects, rowType); + } + + @Override public void implement(Implementor implementor) { + System.out.println(getTable()); + } + + /** + * Override this method so it looks down the tree to find the table this node + * is acting on. + */ + @Override public RelOptTable getTable() { + return getInput().getTable(); + } +} + +// End PigProject.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigRel.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigRel.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRel.java new file mode 100644 index 0000000..868d3d5 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRel.java @@ -0,0 +1,91 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.Convention; +import org.apache.calcite.rel.RelNode; + +import com.google.common.base.Joiner; + +import java.util.ArrayList; +import java.util.List; + +/** + * Relational expression that uses the Pig calling convention. + */ +public interface PigRel extends RelNode { + + /** + * Converts this node to a Pig Latin statement. + */ + void implement(Implementor implementor); + + // String getPigRelationAlias(); + // + // String getFieldName(int index); + + /** Calling convention for relational operations that occur in Pig. */ + Convention CONVENTION = new Convention.Impl("PIG", PigRel.class); + + /** + * Callback for the implementation process that converts a tree of + * {@link PigRel} nodes into complete Pig Latin script. + */ + class Implementor { + + /** + * An ordered list of Pig Latin statements. + * + * <p>See + * <a href="https://pig.apache.org/docs/r0.13.0/start.html#pl-statements";> + * Pig Latin reference</a>. + */ + private final List<String> statements = new ArrayList<>(); + + public String getTableName(RelNode input) { + final List<String> qualifiedName = input.getTable().getQualifiedName(); + return qualifiedName.get(qualifiedName.size() - 1); + } + + public String getPigRelationAlias(RelNode input) { + return getTableName(input); + } + + public String getFieldName(RelNode input, int index) { + return input.getRowType().getFieldList().get(index).getName(); + } + + public void addStatement(String statement) { + statements.add(statement); + } + + public void visitChild(int ordinal, RelNode input) { + assert ordinal == 0; + ((PigRel) input).implement(this); + } + + public List<String> getStatements() { + return statements; + } + + public String getScript() { + return Joiner.on("\n").join(statements); + } + } +} + +// End PigRel.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigRelFactories.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigRelFactories.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRelFactories.java new file mode 100644 index 0000000..8e16cc4 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRelFactories.java @@ -0,0 +1,118 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.Context; +import org.apache.calcite.plan.Contexts; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.AggregateCall; +import org.apache.calcite.rel.core.CorrelationId; +import org.apache.calcite.rel.core.JoinRelType; +import org.apache.calcite.rel.core.RelFactories; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.util.ImmutableBitSet; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Set; + +/** Implementations of factories in {@link RelFactories} + * for the Pig adapter. */ +public class PigRelFactories { + + public static final Context ALL_PIG_REL_FACTORIES = + Contexts.of(PigTableScanFactory.INSTANCE, + PigFilterFactory.INSTANCE, + PigAggregateFactory.INSTANCE, + PigJoinFactory.INSTANCE); + + // prevent instantiation + private PigRelFactories() { + } + + /** + * Implementation of + * {@link org.apache.calcite.rel.core.RelFactories.TableScanFactory} that + * returns a {@link PigTableScan}. + */ + public static class PigTableScanFactory implements RelFactories.TableScanFactory { + + public static final PigTableScanFactory INSTANCE = new PigTableScanFactory(); + + @Override public RelNode createScan(RelOptCluster cluster, RelOptTable table) { + return new PigTableScan(cluster, cluster.traitSetOf(PigRel.CONVENTION), table); + } + } + + /** + * Implementation of + * {@link org.apache.calcite.rel.core.RelFactories.FilterFactory} that + * returns a {@link PigFilter}. + */ + public static class PigFilterFactory implements RelFactories.FilterFactory { + + public static final PigFilterFactory INSTANCE = new PigFilterFactory(); + + @Override public RelNode createFilter(RelNode input, RexNode condition) { + return new PigFilter(input.getCluster(), input.getTraitSet().replace(PigRel.CONVENTION), + input, condition); + } + } + + /** + * Implementation of + * {@link org.apache.calcite.rel.core.RelFactories.AggregateFactory} that + * returns a {@link PigAggregate}. + */ + public static class PigAggregateFactory implements RelFactories.AggregateFactory { + + public static final PigAggregateFactory INSTANCE = new PigAggregateFactory(); + + @Override public RelNode createAggregate(RelNode input, boolean indicator, + ImmutableBitSet groupSet, ImmutableList<ImmutableBitSet> groupSets, + List<AggregateCall> aggCalls) { + return new PigAggregate(input.getCluster(), input.getTraitSet(), input, indicator, groupSet, + groupSets, aggCalls); + } + } + + /** + * Implementation of + * {@link org.apache.calcite.rel.core.RelFactories.JoinFactory} that + * returns a {@link PigJoin}. + */ + public static class PigJoinFactory implements RelFactories.JoinFactory { + + public static final PigJoinFactory INSTANCE = new PigJoinFactory(); + + @Override public RelNode createJoin(RelNode left, RelNode right, RexNode condition, + Set<CorrelationId> variablesSet, JoinRelType joinType, boolean semiJoinDone) { + return new PigJoin(left.getCluster(), left.getTraitSet(), left, right, condition, joinType); + } + + @SuppressWarnings("deprecation") + @Override public RelNode createJoin(RelNode left, RelNode right, RexNode condition, + JoinRelType joinType, Set<String> variablesStopped, boolean semiJoinDone) { + return new PigJoin(left.getCluster(), left.getTraitSet(), left, right, condition, joinType); + } + } +} + +// End PigRelFactories.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigRules.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigRules.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRules.java new file mode 100644 index 0000000..355b6d0 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigRules.java @@ -0,0 +1,144 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.Convention; +import org.apache.calcite.plan.RelOptRule; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; +import org.apache.calcite.rel.logical.LogicalAggregate; +import org.apache.calcite.rel.logical.LogicalFilter; +import org.apache.calcite.rel.logical.LogicalJoin; +import org.apache.calcite.rel.logical.LogicalProject; +import org.apache.calcite.rel.logical.LogicalTableScan; + +import com.google.common.collect.ImmutableList; + +import java.util.List; + +/** + * Various {@link RelOptRule}s using the Pig convention. + */ +public class PigRules { + + public static final List<ConverterRule> ALL_PIG_OPT_RULES = + ImmutableList.of(PigFilterRule.INSTANCE, + PigTableScanRule.INSTANCE, + PigProjectRule.INSTANCE, + PigAggregateRule.INSTANCE, + PigJoinRule.INSTANCE); + + // prevent instantiation + private PigRules() {} + + /** + * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalFilter} to a + * {@link PigFilter}. + */ + private static class PigFilterRule extends ConverterRule { + private static final PigFilterRule INSTANCE = new PigFilterRule(); + + private PigFilterRule() { + super(LogicalFilter.class, Convention.NONE, PigRel.CONVENTION, "PigFilterRule"); + } + + public RelNode convert(RelNode rel) { + final LogicalFilter filter = (LogicalFilter) rel; + final RelTraitSet traitSet = filter.getTraitSet().replace(PigRel.CONVENTION); + return new PigFilter(rel.getCluster(), traitSet, + convert(filter.getInput(), PigRel.CONVENTION), filter.getCondition()); + } + } + + /** + * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalTableScan} + * to a {@link PigTableScan}. + */ + private static class PigTableScanRule extends ConverterRule { + private static final PigTableScanRule INSTANCE = new PigTableScanRule(); + + private PigTableScanRule() { + super(LogicalTableScan.class, Convention.NONE, PigRel.CONVENTION, "PigTableScanRule"); + } + + public RelNode convert(RelNode rel) { + final LogicalTableScan scan = (LogicalTableScan) rel; + final RelTraitSet traitSet = scan.getTraitSet().replace(PigRel.CONVENTION); + return new PigTableScan(rel.getCluster(), traitSet, scan.getTable()); + } + } + + /** + * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalProject} to + * a {@link PigProject}. + */ + private static class PigProjectRule extends ConverterRule { + private static final PigProjectRule INSTANCE = new PigProjectRule(); + + private PigProjectRule() { + super(LogicalProject.class, Convention.NONE, PigRel.CONVENTION, "PigProjectRule"); + } + + public RelNode convert(RelNode rel) { + final LogicalProject project = (LogicalProject) rel; + final RelTraitSet traitSet = project.getTraitSet().replace(PigRel.CONVENTION); + return new PigProject(project.getCluster(), traitSet, project.getInput(), + project.getProjects(), project.getRowType()); + } + } + + /** + * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalAggregate} to a + * {@link PigAggregate}. + */ + private static class PigAggregateRule extends ConverterRule { + private static final PigAggregateRule INSTANCE = new PigAggregateRule(); + + private PigAggregateRule() { + super(LogicalAggregate.class, Convention.NONE, PigRel.CONVENTION, "PigAggregateRule"); + } + + public RelNode convert(RelNode rel) { + final LogicalAggregate agg = (LogicalAggregate) rel; + final RelTraitSet traitSet = agg.getTraitSet().replace(PigRel.CONVENTION); + return new PigAggregate(agg.getCluster(), traitSet, agg.getInput(), + agg.indicator, agg.getGroupSet(), agg.getGroupSets(), agg.getAggCallList()); + } + } + + /** + * Rule to convert a {@link org.apache.calcite.rel.logical.LogicalJoin} to + * a {@link PigJoin}. + */ + private static class PigJoinRule extends ConverterRule { + private static final PigJoinRule INSTANCE = new PigJoinRule(); + + private PigJoinRule() { + super(LogicalJoin.class, Convention.NONE, PigRel.CONVENTION, "PigJoinRule"); + } + + public RelNode convert(RelNode rel) { + final LogicalJoin join = (LogicalJoin) rel; + final RelTraitSet traitSet = join.getTraitSet().replace(PigRel.CONVENTION); + return new PigJoin(join.getCluster(), traitSet, join.getLeft(), join.getRight(), + join.getCondition(), join.getJoinType()); + } + } +} + +// End PigRules.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchema.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchema.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchema.java new file mode 100644 index 0000000..08d5643 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchema.java @@ -0,0 +1,41 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.schema.Table; +import org.apache.calcite.schema.impl.AbstractSchema; + +import java.util.HashMap; +import java.util.Map; + +/** + * Schema that contains one more or more Pig tables. + */ +public class PigSchema extends AbstractSchema { + + protected final Map<String, Table> tableMap = new HashMap<>(); + + @Override protected Map<String, Table> getTableMap() { + return tableMap; + } + + void registerTable(String name, PigTable table) { + tableMap.put(name, table); + } +} + +// End PigSchema.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchemaFactory.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchemaFactory.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchemaFactory.java new file mode 100644 index 0000000..99c91e6 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigSchemaFactory.java @@ -0,0 +1,45 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.SchemaFactory; +import org.apache.calcite.schema.SchemaPlus; + +import java.util.Map; + +/** + * Factory that creates a {@link PigSchema}. + * + * <p>Allows a custom schema to be included in a <code><i>model</i>.json</code> + * file. + */ +public class PigSchemaFactory implements SchemaFactory { + + /** Public singleton, per factory contract. */ + public static final PigSchemaFactory INSTANCE = new PigSchemaFactory(); + + private PigSchemaFactory() { + } + + public Schema create(SchemaPlus parentSchema, String name, + Map<String, Object> operand) { + return new PigSchema(); + } +} + +// End PigSchemaFactory.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigTable.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigTable.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTable.java new file mode 100644 index 0000000..4cfaf10 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTable.java @@ -0,0 +1,75 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelOptTable.ToRelContext; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.schema.TranslatableTable; +import org.apache.calcite.schema.impl.AbstractTable; + +import org.apache.pig.data.DataType; + +/** + * Represents a Pig relation that is created by Pig Latin + * <a href="https://pig.apache.org/docs/r0.13.0/basic.html#load";> + * <code>LOAD</code></a> statement. + * + * <p>Only the default load function is supported at this point (PigStorage()). + * + * <p>Only VARCHAR (CHARARRAY in Pig) type supported at this point. + * + * @see PigTableFactory + */ +public class PigTable extends AbstractTable implements TranslatableTable { + + private final String filePath; + private final String[] fieldNames; + + /** Creates a PigTable. */ + public PigTable(String filePath, String[] fieldNames) { + this.filePath = filePath; + this.fieldNames = fieldNames; + } + + @Override public RelDataType getRowType(RelDataTypeFactory typeFactory) { + final RelDataTypeFactory.FieldInfoBuilder builder = typeFactory.builder(); + for (String fieldName : fieldNames) { + // only supports CHARARRAY types for now + final RelDataType relDataType = typeFactory + .createSqlType(PigDataType.valueOf(DataType.CHARARRAY).getSqlType()); + final RelDataType nullableRelDataType = typeFactory + .createTypeWithNullability(relDataType, true); + builder.add(fieldName, nullableRelDataType); + } + return builder.build(); + } + + public String getFilePath() { + return filePath; + } + + @Override public RelNode toRel(ToRelContext context, RelOptTable relOptTable) { + final RelOptCluster cluster = context.getCluster(); + return new PigTableScan(cluster, cluster.traitSetOf(PigRel.CONVENTION), relOptTable); + } +} + +// End PigTable.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableFactory.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableFactory.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableFactory.java new file mode 100644 index 0000000..51b39cb --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableFactory.java @@ -0,0 +1,55 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.model.ModelHandler; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.TableFactory; + +import java.io.File; +import java.util.List; +import java.util.Map; + +/** + * Factory that creates a {@link PigTable}. + * + * <p>Allows a Pig table to be included in a model.json file.</p> + */ +public class PigTableFactory implements TableFactory<PigTable> { + // public constructor, per factory contract + public PigTableFactory() { + } + + @SuppressWarnings("unchecked") + public PigTable create(SchemaPlus schema, String name, + Map<String, Object> operand, RelDataType rowType) { + String fileName = (String) operand.get("file"); + File file = new File(fileName); + final File base = + (File) operand.get(ModelHandler.ExtraOperand.BASE_DIRECTORY.camelName); + if (base != null && !file.isAbsolute()) { + file = new File(base, fileName); + } + final List<String> fieldNames = (List<String>) operand.get("columns"); + final PigTable result = new PigTable(file.getAbsolutePath(), fieldNames.toArray(new String[0])); + schema.unwrap(PigSchema.class).registerTable(name, result); + return result; + } +} + +// End PigTableFactory.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableScan.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableScan.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableScan.java new file mode 100644 index 0000000..8d5f45e --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigTableScan.java @@ -0,0 +1,92 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.adapter.enumerable.EnumerableRules; +import org.apache.calcite.jdbc.CalciteSchema; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptPlanner; +import org.apache.calcite.plan.RelOptRule; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.core.TableScan; +import org.apache.calcite.rel.rules.AggregateExpandDistinctAggregatesRule; +import org.apache.calcite.rel.type.RelDataTypeField; + +import org.apache.pig.data.DataType; + +import com.google.common.base.Joiner; + +import java.util.ArrayList; +import java.util.List; + +/** Implementation of {@link org.apache.calcite.rel.core.TableScan} in + * {@link PigRel#CONVENTION Pig calling convention}. */ +public class PigTableScan extends TableScan implements PigRel { + + /** Creates a PigTableScan. */ + public PigTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table) { + super(cluster, traitSet, table); + assert getConvention() == PigRel.CONVENTION; + } + + @Override public void implement(Implementor implementor) { + final PigTable pigTable = getPigTable(implementor.getTableName(this)); + final String alias = implementor.getPigRelationAlias(this); + final String schema = '(' + getSchemaForPigStatement(implementor) + + ')'; + final String statement = alias + " = LOAD '" + pigTable.getFilePath() + + "' USING PigStorage() AS " + schema + ';'; + implementor.addStatement(statement); + } + + private PigTable getPigTable(String name) { + final CalciteSchema schema = getTable().unwrap(org.apache.calcite.jdbc.CalciteSchema.class); + return (PigTable) schema.getTable(name, false).getTable(); + } + + private String getSchemaForPigStatement(Implementor implementor) { + final List<String> fieldNamesAndTypes = new ArrayList<>( + getTable().getRowType().getFieldList().size()); + for (RelDataTypeField f : getTable().getRowType().getFieldList()) { + fieldNamesAndTypes.add(getConcatenatedFieldNameAndTypeForPigSchema(implementor, f)); + } + return Joiner.on(", ").join(fieldNamesAndTypes); + } + + private String getConcatenatedFieldNameAndTypeForPigSchema(Implementor implementor, + RelDataTypeField field) { + final PigDataType pigDataType = PigDataType.valueOf(field.getType().getSqlTypeName()); + final String fieldName = implementor.getFieldName(this, field.getIndex()); + return fieldName + ':' + DataType.findTypeName(pigDataType.getPigType()); + } + + @Override public void register(RelOptPlanner planner) { + planner.addRule(PigToEnumerableConverterRule.INSTANCE); + for (RelOptRule rule : PigRules.ALL_PIG_OPT_RULES) { + planner.addRule(rule); + } + // Don't move Aggregates around, otherwise PigAggregate.implement() won't + // know how to correctly procuce Pig Latin + planner.removeRule(AggregateExpandDistinctAggregatesRule.INSTANCE); + // Make sure planner picks PigJoin over EnumerableJoin. Should there be + // a rule for this instead for removing ENUMERABLE_JOIN_RULE here? + planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE); + } +} + +// End PigTableScan.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverter.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverter.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverter.java new file mode 100644 index 0000000..80b8474 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverter.java @@ -0,0 +1,84 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.adapter.enumerable.EnumerableRel; +import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; +import org.apache.calcite.adapter.enumerable.JavaRowFormat; +import org.apache.calcite.adapter.enumerable.PhysType; +import org.apache.calcite.adapter.enumerable.PhysTypeImpl; +import org.apache.calcite.linq4j.tree.BlockBuilder; +import org.apache.calcite.linq4j.tree.Expressions; +import org.apache.calcite.plan.ConventionTraitDef; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterImpl; +import org.apache.calcite.runtime.Hook; +import org.apache.calcite.util.BuiltInMethod; + +import java.util.List; + +/** + * Relational expression representing a scan of a table in a Pig data source. + */ +public class PigToEnumerableConverter + extends ConverterImpl + implements EnumerableRel { + /** Creates a PigToEnumerableConverter. */ + protected PigToEnumerableConverter( + RelOptCluster cluster, + RelTraitSet traits, + RelNode input) { + super(cluster, ConventionTraitDef.INSTANCE, traits, input); + } + + @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) { + return new PigToEnumerableConverter( + getCluster(), traitSet, sole(inputs)); + } + + /** + * {@inheritDoc} + * + * <p>This implementation does not actually execute the associated Pig Latin + * script and return results. Instead it returns an empty + * {@link org.apache.calcite.adapter.enumerable.EnumerableRel.Result} + * in order to allow for testing and verification of every step of query + * processing up to actual physical execution and result verification. + * + * <p>Next step is to invoke Pig from here, likely in local mode, have it + * store results in a predefined file so they can be read here and returned as + * a {@code Result} object. + */ + public Result implement(EnumerableRelImplementor implementor, Prefer pref) { + final BlockBuilder list = new BlockBuilder(); + final PhysType physType = + PhysTypeImpl.of(implementor.getTypeFactory(), rowType, + pref.prefer(JavaRowFormat.ARRAY)); + PigRel.Implementor impl = new PigRel.Implementor(); + impl.visitChild(0, getInput()); + Hook.QUERY_PLAN.run(impl.getScript()); // for script validation in tests + list.add( + Expressions.return_(null, + Expressions.call( + BuiltInMethod.EMPTY_ENUMERABLE.method))); + return implementor.result(physType, list.toBlock()); + } +} + +// End PigToEnumerableConverter.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverterRule.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverterRule.java b/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverterRule.java new file mode 100644 index 0000000..e312186 --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/PigToEnumerableConverterRule.java @@ -0,0 +1,43 @@ +/* + * 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.calcite.adapter.pig; + +import org.apache.calcite.adapter.enumerable.EnumerableConvention; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.convert.ConverterRule; + +/** + * Rule to convert a relational expression from + * {@link PigRel#CONVENTION} to {@link EnumerableConvention}. + */ +public class PigToEnumerableConverterRule extends ConverterRule { + public static final ConverterRule INSTANCE = + new PigToEnumerableConverterRule(); + + private PigToEnumerableConverterRule() { + super(RelNode.class, PigRel.CONVENTION, EnumerableConvention.INSTANCE, + "PigToEnumerableConverterRule"); + } + + @Override public RelNode convert(RelNode rel) { + RelTraitSet newTraitSet = rel.getTraitSet().replace(getOutConvention()); + return new PigToEnumerableConverter(rel.getCluster(), newTraitSet, rel); + } +} + +// End PigToEnumerableConverterRule.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/main/java/org/apache/calcite/adapter/pig/package-info.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/calcite/adapter/pig/package-info.java b/pig/src/main/java/org/apache/calcite/adapter/pig/package-info.java new file mode 100644 index 0000000..9bb298d --- /dev/null +++ b/pig/src/main/java/org/apache/calcite/adapter/pig/package-info.java @@ -0,0 +1,27 @@ +/* + * 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. + */ + +/** + * Pig query provider. + * + */ +@PackageMarker +package org.apache.calcite.adapter.pig; + +import org.apache.calcite.avatica.util.PackageMarker; + +// End package-info.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/test/java/org/apache/calcite/test/AbstractPigTest.java ---------------------------------------------------------------------- diff --git a/pig/src/test/java/org/apache/calcite/test/AbstractPigTest.java b/pig/src/test/java/org/apache/calcite/test/AbstractPigTest.java new file mode 100644 index 0000000..4840ce2 --- /dev/null +++ b/pig/src/test/java/org/apache/calcite/test/AbstractPigTest.java @@ -0,0 +1,36 @@ +/* + * 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.calcite.test; + +import java.io.File; +import java.net.URISyntaxException; + +/** + * Common methods inheritable by all Pig-specific test classes. + */ +public abstract class AbstractPigTest { + + protected String getFullPathForTestDataFile(String fileName) { + try { + return new File(getClass().getResource("/" + fileName).toURI()).getAbsolutePath(); + } catch (URISyntaxException e) { + throw new RuntimeException(e); + } + } +} + +// End AbstractPigTest.java http://git-wip-us.apache.org/repos/asf/calcite/blob/fdbb81cf/pig/src/test/java/org/apache/calcite/test/PigAdapterTest.java ---------------------------------------------------------------------- diff --git a/pig/src/test/java/org/apache/calcite/test/PigAdapterTest.java b/pig/src/test/java/org/apache/calcite/test/PigAdapterTest.java new file mode 100644 index 0000000..16e44f1 --- /dev/null +++ b/pig/src/test/java/org/apache/calcite/test/PigAdapterTest.java @@ -0,0 +1,190 @@ +/* + * 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.calcite.test; + +import com.google.common.base.Function; +import com.google.common.collect.ImmutableMap; + +import org.junit.Test; + +import java.util.List; + +import static org.junit.Assert.assertEquals; + +/** + * Tests for the {@code org.apache.calcite.adapter.pig} package. + */ +public class PigAdapterTest extends AbstractPigTest { + + public static final ImmutableMap<String, String> MODEL = + ImmutableMap.of("model", + PigAdapterTest.class.getResource("/model.json").getPath()); + + @Test + public void testScanAndFilter() throws Exception { + CalciteAssert.that() + .with(MODEL) + .query("select * from \"t\" where \"tc0\" > 'abc'") + .explainContains( + "PigToEnumerableConverter\n" + + " PigFilter(condition=[>($0, 'abc')])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = FILTER t BY (tc0 > 'abc');")); + } + + @Test + public void testImplWithMltipleFilters() { + CalciteAssert.that() + .with(MODEL) + .query("select * from \"t\" where \"tc0\" > 'abc' and \"tc1\" = '3'") + .explainContains( + "PigToEnumerableConverter\n" + + " PigFilter(condition=[AND(>($0, 'abc'), =($1, '3'))])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = FILTER t BY (tc0 > 'abc') AND (tc1 == '3');")); + } + + @Test + public void testImplWithGroupByAndCount() { + CalciteAssert.that() + .with(MODEL) + .query("select count(\"tc1\") c from \"t\" group by \"tc0\"") + .explainContains( + "PigToEnumerableConverter\n" + + " PigAggregate(group=[{0}], C=[COUNT($1)])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = GROUP t BY (tc0);\n" + + "t = FOREACH t {\n" + + " GENERATE group AS tc0, COUNT(t.tc1) AS C;\n" + + "};")); + } + + @Test + public void testImplWithCountWithoutGroupBy() { + CalciteAssert.that() + .with(MODEL) + .query("select count(\"tc0\") c from \"t\"") + .explainContains( + "PigToEnumerableConverter\n" + + " PigAggregate(group=[{}], C=[COUNT($0)])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = GROUP t ALL;\n" + + "t = FOREACH t {\n" + + " GENERATE COUNT(t.tc0) AS C;\n" + + "};")); + } + + @Test + public void testImplWithGroupByMultipleFields() { + CalciteAssert.that() + .with(MODEL) + .query("select * from \"t\" group by \"tc1\", \"tc0\"") + .explainContains( + "PigToEnumerableConverter\n" + + " PigAggregate(group=[{0, 1}])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = GROUP t BY (tc0, tc1);\n" + + "t = FOREACH t {\n" + + " GENERATE group.tc0 AS tc0, group.tc1 AS tc1;\n" + + "};")); + } + + @Test + public void testImplWithGroupByCountDistinct() { + CalciteAssert.that() + .with(MODEL) + .query("select count(distinct \"tc0\") c from \"t\" group by \"tc1\"") + .explainContains( + "PigToEnumerableConverter\n" + + " PigAggregate(group=[{1}], C=[COUNT(DISTINCT $0)])\n" + + " PigTableScan(table=[[PIG, t]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "t = GROUP t BY (tc1);\n" + + "t = FOREACH t {\n" + + " tc0_DISTINCT = DISTINCT t.tc0;\n" + + " GENERATE group AS tc1, COUNT(tc0_DISTINCT) AS C;\n" + + "};")); + } + + @Test + public void testImplWithJoin() throws Exception { + CalciteAssert.that() + .with(MODEL) + .query("select * from \"t\" join \"s\" on \"tc1\"=\"sc0\"") + .explainContains( + "PigToEnumerableConverter\n" + + " PigJoin(condition=[=($1, $2)], joinType=[inner])\n" + + " PigTableScan(table=[[PIG, t]])\n" + + " PigTableScan(table=[[PIG, s]])") + .runs() + .queryContains( + pigScriptChecker( + "t = LOAD '" + getFullPathForTestDataFile("data.txt") + + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n" + + "s = LOAD '" + getFullPathForTestDataFile("data2.txt") + + "' USING PigStorage() AS (sc0:chararray, sc1:chararray);\n" + + "t = JOIN t BY tc1 , s BY sc0;")); + } + + /** Returns a function that checks that a particular Pig Latin scriptis + * generated to implement a query. */ + @SuppressWarnings("rawtypes") + private static Function<List, Void> pigScriptChecker(final String... strings) { + return new Function<List, Void>() { + public Void apply(List actual) { + String actualArray = + actual == null || actual.isEmpty() + ? null + : (String) actual.get(0); + assertEquals("expected Pig script not found", + strings[0], actualArray); + return null; + } + }; + } +} + +// End PigAdapterTest.java
