You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ts...@apache.org on 2013/02/17 20:25:27 UTC
[8/9] Update logical plan to latest spec. Fix existing ROPs to match
new logical spec. Update/add reference storage engine implementations.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/RunningAggregate.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/RunningAggregate.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/RunningAggregate.java
new file mode 100644
index 0000000..01df9a0
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/RunningAggregate.java
@@ -0,0 +1,49 @@
+/*******************************************************************************
+ * 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.drill.common.logical.data;
+
+import org.apache.drill.common.expression.FieldReference;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+@JsonTypeName("runningaggregate")
+public class RunningAggregate extends SingleInputOperator{
+
+ private final FieldReference within;
+ private final NamedExpression[] aggregations;
+
+ @JsonCreator
+ public RunningAggregate(@JsonProperty("within") FieldReference within, @JsonProperty("aggregations") NamedExpression[] aggregations) {
+ super();
+ this.within = within;
+ this.aggregations = aggregations;
+ }
+
+ public FieldReference getWithin() {
+ return within;
+ }
+
+ public NamedExpression[] getAggregations() {
+ return aggregations;
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
index b5333e6..f22a5bc 100644
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
@@ -26,21 +26,21 @@ import com.fasterxml.jackson.annotation.JsonTypeName;
@JsonTypeName("scan")
public class Scan extends SourceOperator{
- private final String sourceName;
+ private final String storageEngine;
private final JSONOptions selection;
private final FieldReference outputReference;
@JsonCreator
- public Scan(@JsonProperty("source") String sourceName, @JsonProperty("selection") JSONOptions selection, @JsonProperty("ref") FieldReference outputReference) {
+ public Scan(@JsonProperty("storageengine") String storageEngine, @JsonProperty("selection") JSONOptions selection, @JsonProperty("ref") FieldReference outputReference) {
super();
- this.sourceName = sourceName;
+ this.storageEngine = storageEngine;
this.selection = selection;
this.outputReference = outputReference;
}
- @JsonProperty("source")
- public String getSourceName() {
- return sourceName;
+ @JsonProperty("storageengine")
+ public String getStorageEngine() {
+ return storageEngine;
}
public JSONOptions getSelection() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Segment.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Segment.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Segment.java
new file mode 100644
index 0000000..5b10e07
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Segment.java
@@ -0,0 +1,47 @@
+/*******************************************************************************
+ * 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.drill.common.logical.data;
+
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.LogicalExpression;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+@JsonTypeName("segment")
+public class Segment extends SingleInputOperator{
+ private final LogicalExpression[] exprs;
+ private final FieldReference name;
+
+ @JsonCreator
+ public Segment(@JsonProperty("exprs") LogicalExpression[] exprs, @JsonProperty("ref") FieldReference name) {
+ super();
+ this.exprs = exprs;
+ this.name = name;
+ }
+
+ public LogicalExpression[] getExprs() {
+ return exprs;
+ }
+
+ public FieldReference getName() {
+ return name;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Sequence.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Sequence.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Sequence.java
index 1c881cc..12c7c08 100644
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Sequence.java
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Sequence.java
@@ -46,9 +46,11 @@ import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
*/
@JsonDeserialize(using = De.class)
@JsonTypeName("sequence")
-public abstract class Sequence extends LogicalOperatorBase {
+public class Sequence extends LogicalOperatorBase {
static final Logger logger = LoggerFactory.getLogger(Sequence.class);
+ private Sequence(){}
+
public boolean openTop;
public LogicalOperator input;
@JsonProperty("do")
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/SinkOperator.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/SinkOperator.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/SinkOperator.java
index 25970ef..08ae075 100644
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/SinkOperator.java
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/SinkOperator.java
@@ -21,7 +21,7 @@ package org.apache.drill.common.logical.data;
/**
* An operator that cannot be subscribed to.
*/
-public class SinkOperator extends SingleInputOperator{
+public abstract class SinkOperator extends SingleInputOperator{
@Override
public void registerAsSubscriber(LogicalOperator operator) {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Store.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Store.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Store.java
new file mode 100644
index 0000000..4168468
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Store.java
@@ -0,0 +1,96 @@
+/*******************************************************************************
+ * 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.drill.common.logical.data;
+
+import org.apache.drill.common.exceptions.ExpressionParsingException;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.logical.JSONOptions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+@JsonTypeName("store")
+public class Store extends SinkOperator{
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Store.class);
+
+ private final String storageEngine;
+ private final JSONOptions target;
+ private final PartitionOptions partition;
+
+ @JsonCreator
+ public Store(@JsonProperty("storageengine") String storageEngine, @JsonProperty("target") JSONOptions target, @JsonProperty("partition") PartitionOptions partition) {
+ super();
+ this.storageEngine = storageEngine;
+ this.target = target;
+ this.partition = partition;
+ }
+
+ public String getStorageEngine() {
+ return storageEngine;
+ }
+
+ public JSONOptions getTarget() {
+ return target;
+ }
+
+ public PartitionOptions getPartition() {
+ return partition;
+ }
+
+ public static enum PartitionType{
+ RANDOM, HASH, ORDERED;
+
+ public static PartitionType resolve(String val){
+ for(PartitionType pt : PartitionType.values()){
+ if(pt.name().equalsIgnoreCase(val)) return pt;
+ }
+ throw new ExpressionParsingException(String.format("Unable to determine partitioning type type for value '%s'.", val));
+
+ }
+
+ };
+
+ public static class PartitionOptions{
+ private final PartitionType partitionType;
+ private final LogicalExpression[] expressions;
+ private final LogicalExpression[] starts;
+
+ @JsonCreator
+ public PartitionOptions(@JsonProperty("partitionType") String partitionType, @JsonProperty("exprs") LogicalExpression[] expressions, @JsonProperty("starts") LogicalExpression[] starts) {
+ this.partitionType = PartitionType.resolve(partitionType);
+ this.expressions = expressions;
+ this.starts = starts;
+ }
+
+ public PartitionType getPartitionType() {
+ return partitionType;
+ }
+
+ public LogicalExpression[] getExpressions() {
+ return expressions;
+ }
+
+ public LogicalExpression[] getStarts() {
+ return starts;
+ }
+
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
index 0153ab9..72a7ffd 100644
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
@@ -23,67 +23,61 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
-@JsonTypeName("window-frame")
-public class WindowFrame {
+@JsonTypeName("windowframe")
+public class WindowFrame extends SingleInputOperator{
- public static final String ALL = "all";
- public static final String HERE = "here";
- private final FieldReference[] keys;
- private final FieldReference ref;
- private final int before;
- private final int after;
+ private final FieldReference within;
+ private final FrameRef frame;
+ private final long start;
+ private final long end;
@JsonCreator
- public WindowFrame(@JsonProperty("keys") FieldReference[] keys, @JsonProperty("ref") FieldReference ref, @JsonProperty("before") String before, @JsonProperty("after") String after) {
+ public WindowFrame(@JsonProperty("within") FieldReference within, @JsonProperty("ref") FrameRef frame, @JsonProperty("start") Long start, @JsonProperty("end") Long end) {
super();
- this.keys = keys;
- this.ref = ref;
- this.before = getVal(before);
- this.after = getVal(after);
- }
-
- private int getVal(String phrase){
- if(ALL.equals(phrase)) return -1;
- if(HERE.equals(phrase)) return 0;
- return Integer.parseInt(phrase);
- }
-
- private String getStr(int val){
- if(val == 0) return HERE;
- if(val == -1) return ALL;
- return Integer.toString(val);
- }
-
- public static String getAll() {
- return ALL;
+ this.within = within;
+ this.frame = frame;
+ this.start = start == null ? Long.MIN_VALUE : start;
+ this.end = end == null ? Long.MIN_VALUE : end;
}
- public static String getHere() {
- return HERE;
+ @JsonProperty("ref")
+ public FrameRef getFrame() {
+ return frame;
}
-
- public FieldReference[] getKeys() {
- return keys;
+ public FieldReference getWithin() {
+ return within;
}
-
- public FieldReference getRef() {
- return ref;
+ public long getStart() {
+ return start;
}
-
- public String getBefore() {
- return getStr(before);
+ public long getEnd() {
+ return end;
}
-
- public String getAfter() {
- return getStr(after);
+ public static class FrameRef{
+ private final FieldReference segment;
+ private final FieldReference position;
+
+ @JsonCreator
+ public FrameRef(@JsonProperty("segment") FieldReference segment, @JsonProperty("position") FieldReference position) {
+ super();
+ this.segment = segment;
+ this.position = position;
+ }
+
+ public FieldReference getSegment() {
+ return segment;
+ }
+ public FieldReference getPosition() {
+ return position;
+ }
+
+
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Write.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Write.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Write.java
deleted file mode 100644
index 4d31732..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Write.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.data;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("write")
-public class Write extends SinkOperator{
- private final String fileName;
-
- @JsonCreator
- public Write(@JsonProperty("file") String fileName) {
- this.fileName = fileName;
- }
-
- public String getFileName() {
- return fileName;
- }
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSource.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSource.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSource.java
deleted file mode 100644
index 6ad5ef8..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSource.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property="type")
-public interface DataSource {
-
- public static final Class<?>[] SUB_TYPES = {JSONDataSource.class, Text.class, Mongo.class, Mysql.class};
-
-
- public String getName();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSourceBase.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSourceBase.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSourceBase.java
deleted file mode 100644
index bce8334..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/DataSourceBase.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-
-public abstract class DataSourceBase implements DataSource{
- public String name;
-
- public String getName() {
- return name;
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/FileBasedDataSource.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/FileBasedDataSource.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/FileBasedDataSource.java
deleted file mode 100644
index 6869123..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/FileBasedDataSource.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import java.util.List;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-public abstract class FileBasedDataSource extends DataSourceBase{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FileBasedDataSource.class);
-
- @JsonProperty("compress") public String compressionType;
- @JsonProperty("files") public List<String> files;
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/JSONDataSource.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/JSONDataSource.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/JSONDataSource.java
deleted file mode 100644
index 7df0dec..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/JSONDataSource.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("json")
-public class JSONDataSource extends FileBasedDataSource{
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mongo.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mongo.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mongo.java
deleted file mode 100644
index 8e11ed0..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mongo.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("mongo")
-public class Mongo extends DataSourceBase{
- public String connection;
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mysql.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mysql.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mysql.java
deleted file mode 100644
index 4141128..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Mysql.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("mysql")
-public class Mysql extends DataSourceBase{
- public String connection;
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Text.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Text.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Text.java
deleted file mode 100644
index 0065c24..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/Text.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources;
-
-import org.apache.drill.common.logical.sources.record.RecordMaker;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("text")
-public class Text extends FileBasedDataSource{
- @JsonProperty("line-delimiter") public String lineDelimiter;
- @JsonProperty("record-maker") public RecordMaker recordMaker;
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/FirstRowMaker.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/FirstRowMaker.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/FirstRowMaker.java
deleted file mode 100644
index 23e86cd..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/FirstRowMaker.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources.record;
-
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-@JsonTypeName("first-row")
-public class FirstRowMaker extends RecordMakerBase{
- public String delimiter;
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMaker.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMaker.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMaker.java
deleted file mode 100644
index 37ec9e0..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMaker.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources.record;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
-public interface RecordMaker {
-
- public static final Class<?>[] SUB_TYPES = { FirstRowMaker.class };
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMakerBase.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMakerBase.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMakerBase.java
deleted file mode 100644
index 33468bc..0000000
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/sources/record/RecordMakerBase.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*******************************************************************************
- * 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.drill.common.logical.sources.record;
-
-public abstract class RecordMakerBase implements RecordMaker{
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/AbstractDynamicBean.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/AbstractDynamicBean.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/AbstractDynamicBean.java
new file mode 100644
index 0000000..5871289
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/AbstractDynamicBean.java
@@ -0,0 +1,81 @@
+/*******************************************************************************
+ * 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.drill.common.util;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.drill.common.exceptions.LogicalPlanParsingException;
+
+import com.fasterxml.jackson.annotation.JsonAnyGetter;
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+import com.fasterxml.jackson.core.JsonParser.Feature;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+/**
+ * Captures all properties and turns them into an object node for late bind conversion.
+ */
+public abstract class AbstractDynamicBean {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractDynamicBean.class);
+
+ private static volatile ObjectMapper MAPPER;
+
+ private ObjectNode objectNode = new ObjectNode(null);
+
+ @JsonAnySetter
+ public void _anySetter(String name, JsonNode value){
+ objectNode.put(name, value);
+ }
+
+ @JsonAnyGetter
+ public Map<String,JsonNode> _anyGetter() {
+ Map<String, JsonNode> unknowns = new HashMap<String, JsonNode>();
+
+ for(Iterator<Entry<String, JsonNode>> i = objectNode.fields(); i.hasNext(); ){
+ Entry<String, JsonNode> e = i.next();
+ unknowns.put(e.getKey(), e.getValue());
+ }
+ return unknowns;
+ }
+
+ public <T> T getWith(Class<T> c){
+ try {
+ return getMapper().treeToValue(objectNode, c);
+ } catch (JsonProcessingException e) {
+ throw new LogicalPlanParsingException(String.format("Failure while trying to convert late bound json type to type of %s.", c.getCanonicalName()), e);
+ }
+ }
+
+
+ private static synchronized ObjectMapper getMapper(){
+ if(MAPPER == null){
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.enable(SerializationFeature.INDENT_OUTPUT);
+ mapper.configure(Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
+ mapper.configure(Feature.ALLOW_COMMENTS, true);
+ MAPPER = mapper;
+ }
+ return MAPPER;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/FileUtils.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/FileUtils.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/FileUtils.java
new file mode 100644
index 0000000..f38f9d2
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/FileUtils.java
@@ -0,0 +1,40 @@
+/*******************************************************************************
+ * 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.drill.common.util;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URL;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+
+public class FileUtils {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FileUtils.class);
+
+ public static File getResourceAsFile(String fileName) throws IOException {
+ URL u = FileUtils.class.getResource(fileName);
+ if (u == null) throw new FileNotFoundException(String.format("Unable to find file on path %s", fileName));
+ return new File(u.getPath());
+ }
+
+ public static String getResourceAsString(String fileName) throws IOException {
+ return Files.toString(getResourceAsFile(fileName), Charsets.UTF_8);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/PathScanner.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/PathScanner.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/PathScanner.java
new file mode 100644
index 0000000..866cc2c
--- /dev/null
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/util/PathScanner.java
@@ -0,0 +1,126 @@
+/*******************************************************************************
+ * 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.drill.common.util;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Modifier;
+import java.net.URL;
+import java.util.Collection;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.drill.common.config.CommonConstants;
+import org.reflections.Reflections;
+import org.reflections.scanners.ResourcesScanner;
+import org.reflections.scanners.SubTypesScanner;
+import org.reflections.scanners.TypeAnnotationsScanner;
+import org.reflections.util.ClasspathHelper;
+import org.reflections.util.ConfigurationBuilder;
+
+import com.google.common.collect.Sets;
+
+public class PathScanner {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PathScanner.class);
+
+ private static final SubTypesScanner subTypeScanner = new SubTypesScanner();
+ private static final TypeAnnotationsScanner annotationsScanner = new TypeAnnotationsScanner();
+ private static final ResourcesScanner resourcesScanner = new ResourcesScanner();
+ private static final Object SYNC = new Object();
+ static volatile Reflections REFLECTIONS = null;
+
+ public static <A extends Annotation, T> Map<A, Class<? extends T>> scanForAnnotatedImplementations(Class<A> annotationClass, Class<T> baseClass, final List<String> scanPackages){
+ Collection<Class<? extends T>> providerClasses = scanForImplementations(baseClass, scanPackages);
+
+ Map<A, Class<? extends T>> map = new HashMap<A, Class<? extends T>>();
+
+ for (Class<? extends T> c : providerClasses) {
+ A annotation = (A) c.getAnnotation(annotationClass);
+ if(annotation == null) continue;
+ map.put(annotation, c);
+ }
+
+ return map;
+ }
+
+ private static Reflections getReflections(){
+ if(REFLECTIONS == null){
+ REFLECTIONS = new Reflections(new ConfigurationBuilder().setUrls(getMarkedPaths()).setScanners(subTypeScanner, annotationsScanner, resourcesScanner));
+ }
+ return REFLECTIONS;
+ }
+
+ public static <T> Class<?>[] scanForImplementationsArr(Class<T> baseClass, final List<String> scanPackages){
+ Collection<Class<? extends T>> imps = scanForImplementations(baseClass, scanPackages);
+ return imps.toArray(new Class<?>[imps.size()]);
+ }
+
+ public static <T> Set<Class<? extends T>> scanForImplementations(Class<T> baseClass, final List<String> scanPackages){
+ synchronized(SYNC){
+ Set<Class<? extends T>> classes = getReflections().getSubTypesOf(baseClass);
+ for(Iterator<Class<? extends T>> i = classes.iterator(); i.hasNext();){
+ Class<? extends T> c = i.next();
+ if(Modifier.isAbstract(c.getModifiers())) i.remove();
+ }
+ return classes;
+ }
+ }
+
+ private static Collection<URL> getMarkedPaths(){
+ return forResource(CommonConstants.DRILL_JAR_MARKER_FILE, true);
+ }
+
+ public static Collection<URL> getConfigURLs(){
+ return forResource(CommonConstants.DRILL_JAR_MARKER_FILE, false);
+ }
+
+ public static Set<URL> forResource(String name, boolean stripName, ClassLoader... classLoaders) {
+ final Set<URL> result = Sets.newHashSet();
+
+ final ClassLoader[] loaders = ClasspathHelper.classLoaders(classLoaders);
+ final String resourceName = name;
+
+ for (ClassLoader classLoader : loaders) {
+ try {
+ final Enumeration<URL> urls = classLoader.getResources(resourceName);
+ while (urls.hasMoreElements()) {
+ final URL url = urls.nextElement();
+
+ int index = url.toExternalForm().lastIndexOf(resourceName);
+ if (index != -1 && stripName) {
+ result.add(new URL(url.toExternalForm().substring(0, index)));
+ } else {
+ result.add(url);
+ }
+ }
+ } catch (IOException e) {
+ if (Reflections.log != null) {
+ Reflections.log.error("error getting resources for package " + name, e);
+ }
+ }
+ }
+
+ return result;
+}
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/resources/drill-default.conf
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/resources/drill-default.conf b/sandbox/prototype/common/src/main/resources/drill-default.conf
new file mode 100644
index 0000000..1b51bfd
--- /dev/null
+++ b/sandbox/prototype/common/src/main/resources/drill-default.conf
@@ -0,0 +1,6 @@
+drill.logical: {
+ operator.packages: ["org.apache.drill.common.logical.data"],
+ expression.packages: ["org.apache.drill.common.expression"],
+ function.packages: ["org.apache.drill.common.expression"],
+ storage.packages: []
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/resources/drill-module.conf b/sandbox/prototype/common/src/main/resources/drill-module.conf
new file mode 100644
index 0000000..28af168
--- /dev/null
+++ b/sandbox/prototype/common/src/main/resources/drill-module.conf
@@ -0,0 +1,3 @@
+// This file tells Drill to consider this module when class path scanning.
+// This file can also include any supplementary configuration information.
+// This file is in HOCON format, see https://github.com/typesafehub/config/blob/master/HOCON.md for more information.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/java/org/apache/drill/storage/CheckStorageConfig.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/java/org/apache/drill/storage/CheckStorageConfig.java b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/CheckStorageConfig.java
new file mode 100644
index 0000000..53fcb9e
--- /dev/null
+++ b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/CheckStorageConfig.java
@@ -0,0 +1,48 @@
+/*******************************************************************************
+ * 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.drill.storage;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collection;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.logical.LogicalPlan;
+import org.apache.drill.common.logical.StorageEngineConfig;
+import org.apache.drill.common.util.FileUtils;
+import org.apache.drill.common.util.PathScanner;
+import org.junit.Test;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+
+public class CheckStorageConfig {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CheckStorageConfig.class);
+
+ @Test
+ public void ensureStorageEnginePickup() {
+ Collection<?> engines = PathScanner.scanForImplementations(StorageEngineConfig.class, Lists.newArrayList("org"));
+ assertEquals(engines.size(), 1);
+ }
+
+ @Test
+ public void checkPlanParsing() throws Exception{
+ LogicalPlan plan = LogicalPlan.parse(DrillConfig.create(), FileUtils.getResourceAsString("/storage_engine_plan.json"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/java/org/apache/drill/storage/MockStorageEngineConfig.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/java/org/apache/drill/storage/MockStorageEngineConfig.java b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/MockStorageEngineConfig.java
new file mode 100644
index 0000000..9f2d9cb
--- /dev/null
+++ b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/MockStorageEngineConfig.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.drill.storage;
+
+import org.apache.drill.common.logical.StorageEngineConfigBase;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+@JsonTypeName("mock")
+public class MockStorageEngineConfig extends StorageEngineConfigBase{
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockStorageEngineConfig.class);
+
+ private String url;
+
+ @JsonCreator
+ public MockStorageEngineConfig(@JsonProperty("name") String name, @JsonProperty("url") String url) {
+ super(name);
+ }
+
+ public String getUrl() {
+ return url;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/java/org/apache/drill/storage/Simple.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/java/org/apache/drill/storage/Simple.java b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/Simple.java
new file mode 100644
index 0000000..a821b59
--- /dev/null
+++ b/sandbox/prototype/common/src/test/java/org/apache/drill/storage/Simple.java
@@ -0,0 +1,33 @@
+/*******************************************************************************
+ * 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.drill.storage;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.logical.LogicalPlan;
+import org.apache.drill.common.util.FileUtils;
+
+public class Simple {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Simple.class);
+
+ public static void main(String[] args) throws Exception {
+ DrillConfig c = DrillConfig.create();
+ LogicalPlan plan = LogicalPlan.parse(c, FileUtils.getResourceAsString("/simple_plan.json"));
+ System.out.println(plan.toJsonString(c));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/resources/drill-module.conf b/sandbox/prototype/common/src/test/resources/drill-module.conf
new file mode 100644
index 0000000..936ed79
--- /dev/null
+++ b/sandbox/prototype/common/src/test/resources/drill-module.conf
@@ -0,0 +1 @@
+drill.logical.storage.packages += "org.apache.drill.storage"
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/resources/simple_plan.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/resources/simple_plan.json b/sandbox/prototype/common/src/test/resources/simple_plan.json
index 86eb57d..070290d 100644
--- a/sandbox/prototype/common/src/test/resources/simple_plan.json
+++ b/sandbox/prototype/common/src/test/resources/simple_plan.json
@@ -7,18 +7,16 @@
info:"na"
}
},
- sources:[
+ storage:[
{
type:"text",
- name:"local-logs",
- files:[
- "local://logs/*.log"
- ],
- compress:"gzip",
- line-delimiter:"\n",
- record-maker:{
- type:"first-row",
- delimiter:","
+ name:"logs",
+ file: "local://logs/*.log",
+ compress:"gzip",
+ line-delimiter:"\n",
+ record-maker:{
+ type:"first-row",
+ delimiter:","
}
},
{
@@ -37,8 +35,8 @@
@id:"1",
op:"scan",
memo:"initial_scan",
- source:"local-logs",
- selection:"activity"
+ storageengine:"local-logs",
+ selection: {}
},
{
@id:"2",
@@ -48,11 +46,11 @@
transforms:[
{
ref:"userId",
- expr:"regex('activity.cookie', \"persistent=([^;]*)\")"
+ expr:"regex_like('activity.cookie', \"persistent=([^;]*)\")"
},
{
ref:"session",
- expr:"regex('activity.cookie', \"session=([^;]*)\")"
+ expr:"regex_like('activity.cookie', \"session=([^;]*)\")"
}
]
},
@@ -64,11 +62,11 @@
transforms:[
{
ref:"userId",
- expr:"regex('activity.cookie', \"persistent=([^;]*)\")"
+ expr:"regex_like('activity.cookie', \"persistent=([^;]*)\")"
},
{
ref:"session",
- expr:"regex('activity.cookie', \"session=([^;]*)\")"
+ expr:"regex_like('activity.cookie', \"session=([^;]*)\")"
}
]
},
@@ -83,7 +81,7 @@
transforms:[
{
ref:"happy",
- expr:"regex('ep2', \"dink\")"
+ expr:"regex_like('ep2', \"dink\")"
}
]
}
@@ -108,33 +106,29 @@
transforms:[
{
ref:"happy",
- expr:"regex('ep2', \"dink\")"
+ expr:"regex_like('ep2', \"dink\")"
}
]
},
{
@id:12,
op:"join",
+ type: "inner",
left:"7",
right:"10",
conditions: [{relationship:"==", left: "1", right: "1" }]
}
,
{
- input:"12",
- op:"project",
- memo:"prj1",
- exprs:[
- {
- ref:"blue",
- expr:"hello"
- },
- {
- ref:"red",
- expr:"users"
- }
- ]
+ input: 12,
+ op: "store",
+ memo: "output sink",
+ target: {
+ file: "console:///stdout"
+ }
+
}
+
]
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/common/src/test/resources/storage_engine_plan.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/test/resources/storage_engine_plan.json b/sandbox/prototype/common/src/test/resources/storage_engine_plan.json
new file mode 100644
index 0000000..e02c481
--- /dev/null
+++ b/sandbox/prototype/common/src/test/resources/storage_engine_plan.json
@@ -0,0 +1,31 @@
+{
+ head:{
+ type:"apache_drill_logical_plan",
+ version:"1",
+ generator:{
+ type:"manual",
+ info:"na"
+ }
+ },
+ storage:[
+ {
+ type:"mock",
+ name:"mock-engine",
+ url: "http://www.apache.org/"
+ }
+ ],
+ query:[
+ {
+ @id:"1",
+ op:"scan",
+ storageengine:"mock-engine",
+ selection: {}
+ },
+ {
+ input: 1,
+ op: "store",
+ storageengine:"mock-engine",
+ target: {}
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/pom.xml
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/pom.xml b/sandbox/prototype/exec/ref/pom.xml
index 0719eb3..85fd68a 100644
--- a/sandbox/prototype/exec/ref/pom.xml
+++ b/sandbox/prototype/exec/ref/pom.xml
@@ -16,49 +16,62 @@
<dep.slf4j.version>1.7.2</dep.slf4j.version>
</properties>
<dependencies>
+
<dependency>
<groupId>org.apache.drill</groupId>
<artifactId>common</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
+
<dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>3.8.1</version>
- <scope>test</scope>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ <version>1.1.0</version>
</dependency>
+
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.0.7</version>
<scope>provided</scope>
</dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>jul-to-slf4j</artifactId>
- <version>${dep.slf4j.version}</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>jcl-over-slf4j</artifactId>
- <version>${dep.slf4j.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>log4j-over-slf4j</artifactId>
- <version>${dep.slf4j.version}</version>
- </dependency>
-
- <dependency>
- <groupId>de.huxhorn.lilith</groupId>
- <artifactId>de.huxhorn.lilith.logback.appender.multiplex-classic</artifactId>
- <version>0.9.42.1</version>
- </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <version>${dep.slf4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <version>${dep.slf4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <version>${dep.slf4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>de.huxhorn.lilith</groupId>
+ <artifactId>de.huxhorn.lilith.logback.appender.multiplex-classic</artifactId>
+ <version>0.9.42.1</version>
+ </dependency>
+
<dependency>
<groupId>com.carrotsearch</groupId>
<artifactId>hppc</artifactId>
<version>0.4.2</version>
</dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.11</version>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ExecRefConstants.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ExecRefConstants.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ExecRefConstants.java
new file mode 100644
index 0000000..bb220df
--- /dev/null
+++ b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ExecRefConstants.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.
+ ******************************************************************************/
+package org.apache.drill.exec.ref;
+
+public class ExecRefConstants {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExecRefConstants.class);
+
+ public static final String CONFIG_DEFAULT = "drill-exec-ref-default.json";
+ public static final String CONFIG_OVERRIDE = "drill-exec-ref-site.json";
+ public static final String STORAGE_ENGINE_SCAN_PACKAGES = "drill.storage.packages";
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/IteratorRegistry.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/IteratorRegistry.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/IteratorRegistry.java
index 24aea0c..6ace788 100644
--- a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/IteratorRegistry.java
+++ b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/IteratorRegistry.java
@@ -52,13 +52,13 @@ public class IteratorRegistry {
}
public List<RecordIterator> getOperator(LogicalOperator o){
- logger.debug("Getting iterator for Logical Operator {}", o);
+// logger.debug("Getting iterator for Logical Operator {}", o);
if(o == null) throw new SetupException("You requested a Iterator list for a null operator. This doesn't make any sense.");
List<ROP> refOps = map.get(o);
List<RecordIterator> iterators = new ArrayList<RecordIterator>(refOps.size());
for(ROP r : refOps){
RecordIterator iterator = r.getOutput();
- if(iterator == null) throw new SetupException(String.format("The provided operator for the reference operator %s is null.", r));
+ if(iterator == null) throw new SetupException(String.format("The provided iterator for the reference operator %s is null.", r));
iterators.add(iterator);
}
return iterators;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ROPConverter.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ROPConverter.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ROPConverter.java
index db9f38b..06a8690 100644
--- a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ROPConverter.java
+++ b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ROPConverter.java
@@ -21,20 +21,25 @@ import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
-import java.util.List;
-import java.util.Queue;
+import java.util.Collection;
import org.apache.drill.common.logical.LogicalPlan;
-import org.apache.drill.common.logical.data.Aggregate;
+import org.apache.drill.common.logical.StorageEngineConfig;
import org.apache.drill.common.logical.data.LogicalOperator;
import org.apache.drill.common.logical.data.Scan;
+import org.apache.drill.common.logical.data.Store;
import org.apache.drill.common.logical.data.Union;
-import org.apache.drill.common.logical.data.Write;
-import org.apache.drill.common.logical.sources.DataSource;
-import org.apache.drill.common.logical.sources.JSONDataSource;
import org.apache.drill.exec.ref.eval.EvaluatorFactory;
import org.apache.drill.exec.ref.exceptions.SetupException;
-import org.apache.drill.exec.ref.rops.*;
+import org.apache.drill.exec.ref.rops.ROP;
+import org.apache.drill.exec.ref.rops.ScanROP;
+import org.apache.drill.exec.ref.rops.StoreROP;
+import org.apache.drill.exec.ref.rops.UnionROP;
+import org.apache.drill.exec.ref.rse.RSERegistry;
+import org.apache.drill.exec.ref.rse.ReferenceStorageEngine;
+import org.apache.drill.exec.ref.rse.ReferenceStorageEngine.ReadEntry;
+
+import com.typesafe.config.Config;
class ROPConverter {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ROPConverter.class);
@@ -42,13 +47,14 @@ class ROPConverter {
private LogicalPlan plan;
private IteratorRegistry registry;
private EvaluatorFactory builder;
- private final List<Queue> sinkQueues;
+ private RSERegistry engineRegistry;
+ private Config config;
- public ROPConverter(LogicalPlan plan, IteratorRegistry registry, EvaluatorFactory builder, List<Queue> sinkQueues) {
+ public ROPConverter(LogicalPlan plan, IteratorRegistry registry, EvaluatorFactory builder, RSERegistry engineRegistry) {
this.plan = plan;
this.registry = registry;
this.builder = builder;
- this.sinkQueues = sinkQueues;
+ this.engineRegistry = engineRegistry;
}
public void convert(LogicalOperator o) throws SetupException {
@@ -68,7 +74,7 @@ class ROPConverter {
if (c instanceof SetupException) {
throw (SetupException) c;
} else {
- throw new RuntimeException("Failure while trying to run convertSpecifc conversion of operator type "
+ throw new RuntimeException("Failure while trying to run convertSpecific conversion of operator type "
+ o.getClass().getSimpleName(), c);
}
}
@@ -98,67 +104,51 @@ class ROPConverter {
+ o.getClass().getCanonicalName());
}
- public void convertSpecific(Write write) throws SetupException {
- if (write.getFileName().startsWith("socket:")) {
- final String ordinalString =
- write.getFileName().substring("socket:".length());
- final Queue queue;
- try {
- int ordinal = Integer.valueOf(ordinalString);
- queue = sinkQueues.get(ordinal);
- } catch (IndexOutOfBoundsException | NumberFormatException e) {
- throw new RuntimeException("bad socket ordinal " + ordinalString);
- }
- final QueueSinkROP queueSink = new QueueSinkROP(write, queue);
- queueSink.init(registry, builder);
- } else {
- JSONWriter writer = new JSONWriter(write);
- writer.init(registry, builder);
- }
+ private ReferenceStorageEngine getEngine(String name){
+ StorageEngineConfig config = plan.getStorageEngine(name);
+ if(config == null) throw new SetupException(String.format("Unable to find define logical plan of name [%s].", name));
+ ReferenceStorageEngine engine = engineRegistry.getEngine(config);
+ return engine;
}
-
- public void convertSpecific(Aggregate agg) throws SetupException {
- if(agg.getAggregateType().equals("simple")){
- ROP r = new AggregateSimpleROP(agg);
- r.init(registry, builder);
- }else{
- throw new SetupException(String.format("Currently, the reference interpreter does not support aggregators of type '%s'. It only supports simple aggregators.", agg.getAggregateType()));
- }
+
+ public void convertSpecific(Store store) throws SetupException {
+ StoreROP rop = new StoreROP(store, getEngine(store.getStorageEngine()));
+ rop.init(registry, builder);
}
public void convertSpecific(Scan scan) throws SetupException {
+ StorageEngineConfig engineConfig = plan.getStorageEngine(scan.getStorageEngine());
+ ReferenceStorageEngine engine = engineRegistry.getEngine(engineConfig);
+ Collection<ReadEntry> readEntries;
try {
- DataSource ds = plan.getDataSource(scan.getSourceName());
-
- if (ds instanceof JSONDataSource) {
- JSONDataSource js = (JSONDataSource) ds;
- List<String> files = js.files;
- if (files.size() > 1) {
- Union logOp = new Union(null);
-
- ROP parentUnion = new UnionROP(logOp);
- ROP[] scanners = new ROP[files.size()];
- for (int i = 0; i < files.size(); i++) {
- scanners[i] = new JSONScanner(scan, files.get(i));
- scanners[i].init(registry, builder);
- }
-
- parentUnion.init(registry, builder);
- registry.swap(logOp, scan); // make it so future things point to
- } else {
- JSONScanner scanner = new JSONScanner(scan, files.get(0));
- scanner.init(registry, builder);
- }
- } else {
- fail(ds);
- }
- } catch (IOException e) {
- throw new SetupException("Failure while trying to set up JSON reading node.", e);
+ readEntries = engine.getReadEntries(scan);
+ } catch (IOException e1) {
+ throw new SetupException("Failure reading input entries.", e1);
+ }
+
+ switch(readEntries.size()){
+ case 0:
+ throw new SetupException(String.format("Scan provided did not correspond to any available data.", scan));
+ case 1:
+ ScanROP scanner = new ScanROP(scan, readEntries.iterator().next(), engine);
+ scanner.init(registry, builder);
+ return;
+ default:
+ Union logOp = new Union(null);
+
+ ROP parentUnion = new UnionROP(logOp);
+ ScanROP[] scanners = new ScanROP[readEntries.size()];
+ int i = 0;
+ for (ReadEntry e : readEntries) {
+ scanners[i] = new ScanROP(scan, e, engine);
+ scanners[i].init(registry, builder);
+ i++;
+ }
+
+ parentUnion.init(registry, builder);
+ registry.swap(logOp, scan); // make it so future things point to the union as the original scans.
+ return;
}
}
- private void fail(Object o) {
- throw new UnsupportedOperationException(String.format(
- "Reference implementation doesn't support interactions with [%s].", o.getClass().getSimpleName()));
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ReferenceInterpreter.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ReferenceInterpreter.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ReferenceInterpreter.java
index 3df6b4b..5be733b 100644
--- a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ReferenceInterpreter.java
+++ b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/ReferenceInterpreter.java
@@ -21,35 +21,31 @@ import java.io.File;
import java.io.IOException;
import java.util.*;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.logical.LogicalPlan;
import org.apache.drill.common.logical.data.LogicalOperator;
-import org.apache.drill.common.logical.sources.DataSource;
-import org.apache.drill.common.logical.sources.record.RecordMaker;
import org.apache.drill.exec.ref.eval.BasicEvaluatorFactory;
import org.apache.drill.exec.ref.eval.EvaluatorFactory;
import org.apache.drill.exec.ref.rops.SinkROP;
+import org.apache.drill.exec.ref.rse.RSERegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.fasterxml.jackson.core.JsonParser.Feature;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
import com.google.common.base.Charsets;
import com.google.common.io.Files;
public class ReferenceInterpreter {
- @SuppressWarnings("unused") private static final Logger logger = LoggerFactory.getLogger(ReferenceInterpreter.class);
+ static final Logger logger = LoggerFactory.getLogger(ReferenceInterpreter.class);
private List<SinkROP> sinks = new ArrayList<SinkROP>();
private LogicalPlan plan;
private ROPConverter converter;
private IteratorRegistry registry;
- public ReferenceInterpreter(LogicalPlan p, IteratorRegistry r,
- EvaluatorFactory builder, List<Queue> sinkQueues){
+ public ReferenceInterpreter(LogicalPlan p, IteratorRegistry r, EvaluatorFactory builder, RSERegistry rses){
this.plan = p;
this.registry = r;
- this.converter = new ROPConverter(p, registry, builder, sinkQueues);
+ this.converter = new ROPConverter(p, registry, builder, rses);
}
/** Generate Reference equivalents to each operation and then collect and store all the sinks.
@@ -58,7 +54,6 @@ public class ReferenceInterpreter {
for(LogicalOperator op : plan.getSortedOperators()){
converter.convert(op);
}
-
sinks.addAll(registry.getSinks());
}
@@ -78,23 +73,11 @@ public class ReferenceInterpreter {
public static void main(String[] args) throws Exception{
-
+ DrillConfig config = DrillConfig.create();
final String jsonFile = args[0];
- ObjectMapper mapper = new ObjectMapper();
-
- mapper.enable(SerializationFeature.INDENT_OUTPUT);
- mapper.configure(Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
- mapper.configure(Feature.ALLOW_COMMENTS, true);
-
- mapper.registerSubtypes(LogicalOperator.SUB_TYPES);
- mapper.registerSubtypes(DataSource.SUB_TYPES);
- mapper.registerSubtypes(RecordMaker.SUB_TYPES);
-
- String externalPlan = Files.toString(new File(jsonFile), Charsets.UTF_8);
- LogicalPlan plan = mapper.readValue(externalPlan, LogicalPlan.class);
+ LogicalPlan plan = LogicalPlan.parse(config, Files.toString(new File(jsonFile), Charsets.UTF_8));
IteratorRegistry ir = new IteratorRegistry();
- ReferenceInterpreter i = new ReferenceInterpreter(plan, ir, new BasicEvaluatorFactory(ir),
- Collections.<Queue>emptyList());
+ ReferenceInterpreter i = new ReferenceInterpreter(plan, ir, new BasicEvaluatorFactory(ir), new RSERegistry(config));
i.setup();
Collection<RunOutcome> outcomes = i.run();
@@ -106,8 +89,6 @@ public class ReferenceInterpreter {
}
}
- //plan.getGraph().getAdjList().printEdges();
- //System.out.println(mapper.writeValueAsString(plan));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AbstractBlockingOperator.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AbstractBlockingOperator.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AbstractBlockingOperator.java
index 6aec21b..57966ce 100644
--- a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AbstractBlockingOperator.java
+++ b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AbstractBlockingOperator.java
@@ -45,10 +45,10 @@ public abstract class AbstractBlockingOperator<T extends SingleInputOperator> ex
@Override
public NextOutcome next() {
- if(iter == null){
- iter = consumeData();
+ if(this.iter == null){
+ this.iter = consumeData();
}
- return iter.next();
+ return this.iter.next();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/6e4a1384/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AggregateSimpleROP.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AggregateSimpleROP.java b/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AggregateSimpleROP.java
deleted file mode 100644
index e059688..0000000
--- a/sandbox/prototype/exec/ref/src/main/java/org/apache/drill/exec/ref/rops/AggregateSimpleROP.java
+++ /dev/null
@@ -1,219 +0,0 @@
-/*******************************************************************************
- * 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.drill.exec.ref.rops;
-
-import org.apache.drill.common.expression.FieldReference;
-import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.common.logical.data.Aggregate;
-import org.apache.drill.exec.ref.RecordIterator;
-import org.apache.drill.exec.ref.RecordPointer;
-import org.apache.drill.exec.ref.UnbackedRecord;
-import org.apache.drill.exec.ref.eval.EvaluatorFactory;
-import org.apache.drill.exec.ref.eval.EvaluatorTypes.AggregatingEvaluator;
-import org.apache.drill.exec.ref.eval.EvaluatorTypes.BasicEvaluator;
-import org.apache.drill.exec.ref.values.DataValue;
-
-public class AggregateSimpleROP extends SingleInputROPBase<Aggregate> implements BoundaryListener {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AggregateSimpleROP.class);
-
- private AggregatingEvaluator[] eval;
- private SchemaPath[] outputNames;
- private boolean boundaryCrossed = false;
- private RecordIterator incoming;
- private FieldReference[] orderingBoundaries;
- private DataValue[] previousValues;
- private DataValue[] currentValues;
- private BasicEvaluator[] boundaryKeys;
-
- public AggregateSimpleROP(Aggregate config) {
- super(config);
- int len = config.getKeys().length;
- orderingBoundaries = new FieldReference[len];
- previousValues = new DataValue[len];
- currentValues = new DataValue[len];
- boundaryKeys = new BasicEvaluator[len];
-
- for(int i = 0; i < orderingBoundaries.length; i++){
- orderingBoundaries[i] = config.getKeys()[i];
- }
- }
-
- @Override
- protected void setupEvals(EvaluatorFactory builder) {
- eval = new AggregatingEvaluator[config.getAggregations().length];
- outputNames = new SchemaPath[eval.length];
- for(int i =0; i < eval.length; i++){
- eval[i] = builder.getAggregatingOperator(record, config.getAggregations()[i].getExpr());
- outputNames[i] = config.getAggregations()[i].getRef();
- }
-
- for(int i =0; i < orderingBoundaries.length; i++){
- boundaryKeys[i] = builder.getBasicEvaluator(record, orderingBoundaries[i]);
- }
-
- }
-
- @Override
- protected void setInput(RecordIterator incoming) {
- this.incoming = incoming;
- }
-
- @Override
- protected RecordIterator getIteratorInternal() {
- return new AggregatingIterator();
- }
-
- private boolean checkBoundaryCrossing() {
- if (boundaryCrossed) {
- boundaryCrossed = false;
- return true;
- } else {
- return false;
- }
- }
-
- @Override
- public void markBoundaryAsCrossed() {
- this.boundaryCrossed = true;
- }
-
-
-
-
- private class AggregatingIterator implements RecordIterator {
- private boolean remainder = false;
- private boolean more = true;
- private UnbackedRecord outputRecord = new UnbackedRecord();
- public AggregatingIterator() {
- }
-
- private NextOutcome readNext(){
-
- // copy over the previous values.
- System.arraycopy(currentValues, 0, previousValues, 0, currentValues.length);
-
- // increment the parent forward.
- NextOutcome n = incoming.next();
-
- boolean changed = false;
-
- // read boundary values unless no new values were loaded.
- if(n != NextOutcome.NONE_LEFT){
- // collect up the current values for each boundary key.
- for(int i =0; i < currentValues.length; i++){
- currentValues[i] = boundaryKeys[i].eval();
-
- // set a change if the boundary value doesn't equal the previous boundary value
- if(!currentValues[i].equals(previousValues[i])) changed = true;
- }
-
- // skip first boundary.
- if(previousValues[0] == null) changed = false;
-
- }else{
-
- changed = true;
- }
-
- if(changed){
-// logger.debug("Boundary checked, found a crossed boundary for current values: {}, previous values: {}", currentValues, previousValues);
- markBoundaryAsCrossed();
- }else{
-// logger.debug("Boundary checked, found no crossed boundary for current values: {}, previous values: {}", currentValues, previousValues);
-
- }
-
- return n;
- }
-
- private void consumeCurrent(){
- for(int x = 0; x < eval.length; x++){
- eval[x].addRecord();
- }
- }
-
- private void writeOutputRecord(){
- outputRecord.clear();
- for(int x = 0; x < eval.length; x++){
- DataValue dv = eval[x].eval();
-// logger.debug("Adding Aggregated Values named {} with value {}", outputNames[x], dv);
- outputRecord.addField(outputNames[x], dv);
- }
-
- // Add the grouping keys.
- for(int y = 0; y < currentValues.length; y++){
- //logger.debug("Adding Grouping Keys {}", orderingBoundaries[y]);
- outputRecord.addField(orderingBoundaries[y], previousValues[y]);
- }
- }
-
- @Override
- public NextOutcome next() {
-
- NextOutcome whatNext = null;
-
- // if we don't have more and there are no more values, exit.
- if (!more) return NextOutcome.NONE_LEFT;
-
- while (true) {
-
- // we shouldn't increment the iterator since we have to consume our remainder.
- if(!remainder){
- whatNext = readNext();
-
- // if we've just crossed a boundary, we should output the previous values.
- if(checkBoundaryCrossing()){
- writeOutputRecord();
-
- // if there is no future input, we'll flag as !more. Otherwise, will inform that there are pending records that should be consumed.
- if(whatNext == NextOutcome.NONE_LEFT){
- more = false;
- }else{
- remainder = true;
- }
-
- // always return a next outcome of true since we've just output a record.
- return NextOutcome.INCREMENTED_SCHEMA_CHANGED;
-
- }
- }
-
- // we don't consume the current record until after we've output a record associated with a boundary (as necessary).
- consumeCurrent();
- remainder = false;
- }
- }
-
- @Override
- public ROP getParent() {
- return AggregateSimpleROP.this;
- }
-
- @Override
- public RecordPointer getRecordPointer() {
- return outputRecord;
- }
-
- }
-
- @Override
- public SchemaPath[] getOrderedBoundaryPaths() {
- return this.orderingBoundaries;
- }
-
-}