You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by vo...@apache.org on 2018/10/08 12:43:50 UTC
[drill] branch master updated (3b1ae15 -> a253391)
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git.
from 3b1ae15 DRILL-6755: Avoid building Hash Table for inner/left join when probe side is empty
new 9d3c4f7 DRILL-6765: Exclude unused shaded guava classes from drill-jdbc-all jar
new 0ca84ea DRILL-6759: Make columns array name for csv data case insensitive
new 2bd26ff DRILL-6762: Fix dynamic UDFs versioning issue
new 778e043 DRILL-6764: Query fails with IOB when Unnest has reference to deep nested field like (t.c_orders.o_lineitems).
new d5bafec DRILL-6773: The renamed schema with aliases is not shown for queries on empty directories
new a253391 DRILL-6541: Upgrade ZooKeeper patch version to 3.4.11 for mapr profile
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../drill/common/exceptions/ErrorHelper.java | 13 ++++
.../common/util/function/CheckedFunction.java | 63 ++++++++++++++++++
.../common/util/function}/CheckedSupplier.java | 4 +-
.../common/util/function/TestCheckedFunction.java | 35 +++++++---
.../store/mapr/db/MapRDBPushFilterIntoScan.java | 4 +-
contrib/storage-hbase/pom.xml | 1 +
.../exec/store/hbase/HBasePushFilterIntoScan.java | 2 +-
.../store/kafka/KafkaPushDownFilterIntoScan.java | 2 +-
.../store/mongo/MongoPushDownFilterForScan.java | 2 +-
.../exec/coord/store/TransientStoreFactory.java | 2 +-
.../expr/fn/FunctionImplementationRegistry.java | 51 +++++++++------
.../expr/fn/registry/FunctionRegistryHolder.java | 55 ++++++++--------
.../expr/fn/registry/LocalFunctionRegistry.java | 20 +++---
.../expr/fn/registry/RemoteFunctionRegistry.java | 43 +++++++------
.../drill/exec/physical/base/SchemalessScan.java | 6 ++
.../physical/impl/project/ProjectRecordBatch.java | 2 +-
.../planner/physical/ConvertCountToDirectScan.java | 2 +-
.../exec/planner/physical/ProjectAllowDupPrel.java | 14 ++--
.../drill/exec/planner/physical/ProjectPrel.java | 2 +-
.../drill/exec/planner/physical/ScanPrel.java | 12 ++--
.../drill/exec/planner/physical/ScanPrule.java | 2 +-
.../physical/visitor/StarColumnConverter.java | 26 +++++---
.../drill/exec/planner/sql/DrillOperatorTable.java | 6 +-
.../drill/exec/planner/sql/DrillSqlWorker.java | 11 ++--
.../planner/sql/handlers/ComplexUnnestVisitor.java | 11 ++--
.../planner/sql/handlers/DefaultSqlHandler.java | 1 +
.../easy/text/compliant/RepeatedVarCharOutput.java | 2 +-
.../InfoSchemaPushFilterIntoRecordGenerator.java | 2 +-
.../exec/store/parquet/ParquetPushDownFilter.java | 2 +-
.../exec/store/sys/PersistentStoreProvider.java | 23 +++++--
.../exec/store/sys/store/DataChangeVersion.java | 10 ++-
.../UndefinedVersionDelegatingStore.java} | 49 +++++++-------
.../store/sys/store/VersionedDelegatingStore.java | 14 ++--
.../provider/CachingPersistentStoreProvider.java | 51 ++++++++-------
.../sys/store/provider/InMemoryStoreProvider.java | 12 +++-
.../provider/LocalPersistentStoreProvider.java | 6 ++
.../test/java/org/apache/drill/TestUnionAll.java | 4 +-
.../java/org/apache/drill/TestUnionDistinct.java | 4 +-
.../org/apache/drill/exec/TestEmptyInputSql.java | 37 +++++++++--
.../fn/registry/FunctionRegistryHolderTest.java | 75 ++++++++++------------
.../impl/lateraljoin/TestLateralPlans.java | 20 ++++++
.../drill/exec/store/text/TestTextColumn.java | 23 ++++++-
.../exec/udf/dynamic/TestDynamicUDFSupport.java | 18 +++---
.../java/org/apache/drill/test/PrintingUtils.java | 2 +-
...sted-customer.json => nested-customer-map.json} | 28 ++++++--
exec/jdbc-all/pom.xml | 34 +++++-----
pom.xml | 2 +-
47 files changed, 533 insertions(+), 277 deletions(-)
create mode 100644 common/src/main/java/org/apache/drill/common/util/function/CheckedFunction.java
rename {exec/java-exec/src/main/java/org/apache/drill/exec/util => common/src/main/java/org/apache/drill/common/util/function}/CheckedSupplier.java (92%)
copy exec/interpreter/src/test/java/org/apache/drill/exec/expr/TestPrune.java => common/src/test/java/org/apache/drill/common/util/function/TestCheckedFunction.java (51%)
copy exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/{CaseInsensitivePersistentStore.java => store/UndefinedVersionDelegatingStore.java} (51%)
copy exec/java-exec/src/test/resources/lateraljoin/{nested-customer.json => nested-customer-map.json} (82%)
[drill] 01/06: DRILL-6765: Exclude unused shaded guava classes from
drill-jdbc-all jar
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 9d3c4f740297d2cb03165b02182bafa7021d48f9
Author: Igor Guzenko <ih...@gmail.com>
AuthorDate: Tue Oct 2 20:45:11 2018 +0300
DRILL-6765: Exclude unused shaded guava classes from drill-jdbc-all jar
closes #1486
---
exec/jdbc-all/pom.xml | 34 ++++++++++++++++------------------
1 file changed, 16 insertions(+), 18 deletions(-)
diff --git a/exec/jdbc-all/pom.xml b/exec/jdbc-all/pom.xml
index 7a6c9ee..cdd5dc9 100644
--- a/exec/jdbc-all/pom.xml
+++ b/exec/jdbc-all/pom.xml
@@ -436,15 +436,14 @@
<exclude>codegen/**</exclude>
<exclude>bootstrap-storage-plugins.json</exclude>
<exclude>org/apache/parquet</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/math</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/graph</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/net</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/primitives</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/reflect</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/util</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/cache</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/collect/Tree*</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/collect/Standard*</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/escape/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/eventbus/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/html/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/net/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/xml/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/graph/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/collect/Tree*</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/collect/Standard*</exclude>
<exclude>com/google/common/math</exclude>
<exclude>com/google/common/net</exclude>
<exclude>com/google/common/primitives</exclude>
@@ -751,15 +750,14 @@
<exclude>codegen/**</exclude>
<exclude>bootstrap-storage-plugins.json</exclude>
<exclude>org/apache/parquet</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/math</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/graph</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/net</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/primitives</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/reflect</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/util</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/cache</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/collect/Tree*</exclude>
- <exclude>org/apache/drill/shaded/com/google/common/collect/Standard*</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/escape/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/eventbus/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/html/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/net/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/xml/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/graph/**</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/collect/Tree*</exclude>
+ <exclude>org/apache/drill/shaded/guava/com/google/common/collect/Standard*</exclude>
<exclude>com/google/common/math</exclude>
<exclude>com/google/common/net</exclude>
<exclude>com/google/common/primitives</exclude>
[drill] 02/06: DRILL-6759: Make columns array name for csv data
case insensitive
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 0ca84ea40812527b481c5a052687021b43fdfc88
Author: Arina Ielchiieva <ar...@gmail.com>
AuthorDate: Tue Oct 2 16:48:53 2018 +0300
DRILL-6759: Make columns array name for csv data case insensitive
closes #1485
---
.../easy/text/compliant/RepeatedVarCharOutput.java | 2 +-
.../drill/exec/store/text/TestTextColumn.java | 23 +++++++++++++++++++---
2 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
index 57e26f2..05f0773 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
@@ -133,7 +133,7 @@ class RepeatedVarCharOutput extends TextOutput {
for (SchemaPath path : columns) {
assert path.getRootSegment().isNamed() : "root segment should be named";
pathStr = path.getRootSegment().getPath();
- Preconditions.checkArgument(COL_NAME.equals(pathStr) || (SchemaPath.DYNAMIC_STAR.equals(pathStr) && path.getRootSegment().getChild() == null),
+ Preconditions.checkArgument(COL_NAME.equalsIgnoreCase(pathStr) || (SchemaPath.DYNAMIC_STAR.equals(pathStr) && path.getRootSegment().getChild() == null),
String.format("Selected column '%s' must have name 'columns' or must be plain '*'", pathStr));
if (path.getRootSegment().getChild() != null) {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestTextColumn.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestTextColumn.java
index 8dd122b..2b96924 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestTextColumn.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestTextColumn.java
@@ -22,7 +22,8 @@ import java.util.List;
import org.apache.drill.test.BaseTestQuery;
import org.apache.drill.exec.rpc.user.QueryDataBatch;
import org.junit.Test;
-import static org.junit.Assert.assertTrue;
+
+import static org.junit.Assert.assertEquals;
public class TestTextColumn extends BaseTestQuery {
@Test
@@ -41,7 +42,7 @@ public class TestTextColumn extends BaseTestQuery {
expectedResultSet.addRow("g, h,\",\"i\",\"j,, \\n k");
TestResultSet actualResultSet = new TestResultSet(actualResults);
- assertTrue(expectedResultSet.equals(actualResultSet));
+ assertEquals(expectedResultSet, actualResultSet);
}
@Test
@@ -55,6 +56,22 @@ public class TestTextColumn extends BaseTestQuery {
expectedResultSet.addRow("g, h,", "i", "j,, \\n k", "l\\\"m");
TestResultSet actualResultSet = new TestResultSet(actualResults);
- assertTrue(expectedResultSet.equals(actualResultSet));
+ assertEquals(expectedResultSet, actualResultSet);
+ }
+
+ @Test
+ public void testColumnsCaseInsensitive() throws Exception {
+ testBuilder()
+ .sqlQuery("select columns as c from cp.`store/text/data/letters.csv`")
+ .unOrdered()
+ .sqlBaselineQuery("select COLUMNS as c from cp.`store/text/data/letters.csv`")
+ .go();
+
+ testBuilder()
+ .sqlQuery("select columns[0], columns[1] from cp.`store/text/data/letters.csv`")
+ .unOrdered()
+ .sqlBaselineQuery("select COLUMNS[0], CoLuMnS[1] from cp.`store/text/data/letters.csv`")
+ .go();
}
+
}
[drill] 03/06: DRILL-6762: Fix dynamic UDFs versioning issue
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 2bd26ffe34108f876184c63188122f2d49db4c4a
Author: Arina Ielchiieva <ar...@gmail.com>
AuthorDate: Fri Sep 28 20:47:40 2018 +0200
DRILL-6762: Fix dynamic UDFs versioning issue
1. Added UndefinedVersionDelegatingStore to serve as versioned wrapper for those stores that do not support versioning.
2. Aligned remote and local function registries version type. Type will be represented as int since ZK version is returned as int.
3. Added NOT_AVAILABLE and UNDEFINED versions to DataChangeVersion holder to indicate proper registry state.
4. Added additional trace logging.
5. Minor refactoring and clean up.
closes #1484
---
.../drill/common/exceptions/ErrorHelper.java | 13 ++++
.../common/util/function/CheckedFunction.java | 63 +++++++++++++++++
.../common/util/function}/CheckedSupplier.java | 4 +-
.../common/util/function/TestCheckedFunction.java | 53 ++++++++++++++
.../exec/coord/store/TransientStoreFactory.java | 2 +-
.../expr/fn/FunctionImplementationRegistry.java | 51 +++++++++-----
.../expr/fn/registry/FunctionRegistryHolder.java | 55 +++++++--------
.../expr/fn/registry/LocalFunctionRegistry.java | 20 +++---
.../expr/fn/registry/RemoteFunctionRegistry.java | 43 ++++++------
.../drill/exec/planner/sql/DrillOperatorTable.java | 6 +-
.../drill/exec/planner/sql/DrillSqlWorker.java | 11 +--
.../exec/store/sys/PersistentStoreProvider.java | 23 ++++--
.../exec/store/sys/store/DataChangeVersion.java | 10 ++-
.../sys/store/UndefinedVersionDelegatingStore.java | 82 ++++++++++++++++++++++
.../store/sys/store/VersionedDelegatingStore.java | 14 ++--
.../provider/CachingPersistentStoreProvider.java | 51 +++++++-------
.../sys/store/provider/InMemoryStoreProvider.java | 12 +++-
.../provider/LocalPersistentStoreProvider.java | 6 ++
.../fn/registry/FunctionRegistryHolderTest.java | 75 +++++++++-----------
.../exec/udf/dynamic/TestDynamicUDFSupport.java | 18 ++---
.../java/org/apache/drill/test/PrintingUtils.java | 2 +-
21 files changed, 438 insertions(+), 176 deletions(-)
diff --git a/common/src/main/java/org/apache/drill/common/exceptions/ErrorHelper.java b/common/src/main/java/org/apache/drill/common/exceptions/ErrorHelper.java
index 2a331bb..d2dc0f8 100644
--- a/common/src/main/java/org/apache/drill/common/exceptions/ErrorHelper.java
+++ b/common/src/main/java/org/apache/drill/common/exceptions/ErrorHelper.java
@@ -181,4 +181,17 @@ public class ErrorHelper {
return (UserException) cause;
}
+ /**
+ * Helps to hide checked exception from the compiler but then actually throw it.
+ * Is useful when implementing functional interfaces that allow checked exceptions.
+ *
+ * @param e original exception instance
+ * @param <E> exception type
+ * @throws E exception instance
+ */
+ @SuppressWarnings("unchecked")
+ public static <E extends Throwable> void sneakyThrow(Throwable e) throws E {
+ throw (E) e;
+ }
+
}
diff --git a/common/src/main/java/org/apache/drill/common/util/function/CheckedFunction.java b/common/src/main/java/org/apache/drill/common/util/function/CheckedFunction.java
new file mode 100644
index 0000000..050565e
--- /dev/null
+++ b/common/src/main/java/org/apache/drill/common/util/function/CheckedFunction.java
@@ -0,0 +1,63 @@
+/*
+ * 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.function;
+
+import java.util.function.Function;
+
+import static org.apache.drill.common.exceptions.ErrorHelper.sneakyThrow;
+
+/**
+ * Extension of {@link Function} that allows to throw checked exception.
+ *
+ * @param <T> function argument type
+ * @param <R> function result type
+ * @param <E> exception type
+ */
+@FunctionalInterface
+public interface CheckedFunction<T, R, E extends Throwable> extends Function<T, R> {
+
+ /**
+ * Overrides {@link Function#apply(Object)} method to allow calling functions that throw checked exceptions.
+ * Is useful when used in methods that accept {@link Function}.
+ * For example: {@link java.util.Map#computeIfAbsent(Object, Function)}.
+ *
+ * @param t the function argument
+ * @return the function result
+ */
+ @Override
+ default R apply(T t) {
+ try {
+ return applyAndThrow(t);
+ } catch (Throwable e) {
+ sneakyThrow(e);
+ }
+ // should never happen
+ throw new RuntimeException();
+ }
+
+ /**
+ * Applies function to the given argument.
+ *
+ * @param t the function argument
+ * @return the function result
+ * @throws E exception in case of errors
+ */
+ R applyAndThrow(T t) throws E;
+
+}
+
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/CheckedSupplier.java b/common/src/main/java/org/apache/drill/common/util/function/CheckedSupplier.java
similarity index 92%
rename from exec/java-exec/src/main/java/org/apache/drill/exec/util/CheckedSupplier.java
rename to common/src/main/java/org/apache/drill/common/util/function/CheckedSupplier.java
index b744ac8..6063383 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/CheckedSupplier.java
+++ b/common/src/main/java/org/apache/drill/common/util/function/CheckedSupplier.java
@@ -15,10 +15,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.util;
+package org.apache.drill.common.util.function;
/**
- * The java standard library does not provide a lambda function interface for funtions that take no arguments,
+ * The java standard library does not provide a lambda function interface for functions that take no arguments,
* but that throw an exception. So, we have to define our own here.
* @param <T> The return type of the lambda function.
* @param <E> The type of exception thrown by the lambda function.
diff --git a/common/src/test/java/org/apache/drill/common/util/function/TestCheckedFunction.java b/common/src/test/java/org/apache/drill/common/util/function/TestCheckedFunction.java
new file mode 100644
index 0000000..a1ab389
--- /dev/null
+++ b/common/src/test/java/org/apache/drill/common/util/function/TestCheckedFunction.java
@@ -0,0 +1,53 @@
+/*
+ * 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.function;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class TestCheckedFunction {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void testComputeIfAbsentWithCheckedFunction() {
+ ExceptionProducer producer = new ExceptionProducer();
+ Map<String, String> map = new HashMap<>();
+ String message = "Exception message";
+ CheckedFunction<String, String, Exception> function = producer::failWithMessage;
+
+ thrown.expect(Exception.class);
+ thrown.expectMessage(message);
+
+ map.computeIfAbsent(message, function);
+ }
+
+ private class ExceptionProducer {
+
+ String failWithMessage(String message) throws Exception {
+ throw new Exception(message);
+ }
+
+ }
+
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/store/TransientStoreFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/store/TransientStoreFactory.java
index 135ccd4..35feaa9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/coord/store/TransientStoreFactory.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/coord/store/TransientStoreFactory.java
@@ -25,7 +25,7 @@ public interface TransientStoreFactory extends AutoCloseable {
/**
* Returns a {@link TransientStore transient store} instance for the given configuration.
*
- * Note that implementors have liberty to cache previous {@link PersistentStore store} instances.
+ * Note that implementors have liberty to cache previous {@link TransientStore store} instances.
*
* @param config store configuration
* @param <V> store value type
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
index f24f9aa..f4b8373 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/FunctionImplementationRegistry.java
@@ -25,11 +25,12 @@ import java.net.JarURLConnection;
import java.net.URL;
import java.net.URLClassLoader;
import java.net.URLConnection;
+import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
import org.apache.drill.shaded.guava.com.google.common.collect.Sets;
@@ -64,7 +65,6 @@ import org.apache.drill.exec.server.options.OptionSet;
import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
-import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
import org.apache.drill.exec.store.sys.store.DataChangeVersion;
import org.apache.drill.exec.util.JarUtil;
import org.apache.hadoop.fs.FileSystem;
@@ -83,7 +83,7 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
private final Path localUdfDir;
private boolean deleteTmpDir = false;
private File tmpDir;
- private List<PluggableFunctionRegistry> pluggableFuncRegistries = Lists.newArrayList();
+ private List<PluggableFunctionRegistry> pluggableFuncRegistries = new ArrayList<>();
private OptionSet optionManager;
private final boolean useDynamicUdfs;
@@ -168,7 +168,7 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
*/
@Override
public DrillFuncHolder findDrillFunction(FunctionResolver functionResolver, FunctionCall functionCall) {
- AtomicLong version = new AtomicLong();
+ AtomicInteger version = new AtomicInteger();
String newFunctionName = functionReplacement(functionCall);
// Dynamic UDFS: First try with exact match. If not found, we may need to
@@ -246,7 +246,7 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
List<MajorType> argTypes,
MajorType returnType,
boolean retry) {
- AtomicLong version = new AtomicLong();
+ AtomicInteger version = new AtomicInteger();
for (DrillFuncHolder h : localFunctionRegistry.getMethods(name, version)) {
if (h.matches(returnType, argTypes)) {
return h;
@@ -321,19 +321,19 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
/**
* Purpose of this method is to synchronize remote and local function registries if needed
* and to inform if function registry was changed after given version.
- *
+ * <p/>
* To make synchronization as much light-weigh as possible, first only versions of both registries are checked
* without any locking. If synchronization is needed, enters synchronized block to prevent others loading the same jars.
* The need of synchronization is checked again (double-check lock) before comparing jars.
* If any missing jars are found, they are downloaded to local udf area, each is wrapped into {@link JarScan}.
* Once jar download is finished, all missing jars are registered in one batch.
* In case if any errors during jars download / registration, these errors are logged.
- *
+ * <p/>
* During registration local function registry is updated with remote function registry version it is synced with.
* When at least one jar of the missing jars failed to download / register,
* local function registry version are not updated but jars that where successfully downloaded / registered
* are added to local function registry.
- *
+ * <p/>
* If synchronization between remote and local function registry was not needed,
* checks if given registry version matches latest sync version
* to inform if function registry was changed after given version.
@@ -342,16 +342,16 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
* @return true if remote and local function registries were synchronized after given version
*/
@SuppressWarnings("resource")
- public boolean syncWithRemoteRegistry(long version) {
+ public boolean syncWithRemoteRegistry(int version) {
// Do the version check only if a remote registry exists. It does
// not exist for some JMockit-based unit tests.
if (isRegistrySyncNeeded()) {
synchronized (this) {
- long localRegistryVersion = localFunctionRegistry.getVersion();
+ int localRegistryVersion = localFunctionRegistry.getVersion();
if (isRegistrySyncNeeded(remoteFunctionRegistry.getRegistryVersion(), localRegistryVersion)) {
DataChangeVersion remoteVersion = new DataChangeVersion();
List<String> missingJars = getMissingJars(this.remoteFunctionRegistry, localFunctionRegistry, remoteVersion);
- List<JarScan> jars = Lists.newArrayList();
+ List<JarScan> jars = new ArrayList<>();
if (!missingJars.isEmpty()) {
logger.info("Starting dynamic UDFs lazy-init process.\n" +
"The following jars are going to be downloaded and registered locally: " + missingJars);
@@ -381,7 +381,7 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
}
}
}
- long latestRegistryVersion = jars.size() != missingJars.size() ?
+ int latestRegistryVersion = jars.size() != missingJars.size() ?
localRegistryVersion : remoteVersion.getVersion();
localFunctionRegistry.register(jars, latestRegistryVersion);
return true;
@@ -392,23 +392,38 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
return version != localFunctionRegistry.getVersion();
}
+ /**
+ * Checks if remote and local registries should be synchronized.
+ * Before comparing versions, checks if remote function registry is actually exists.
+ *
+ * @return true is local registry should be refreshed, false otherwise
+ */
private boolean isRegistrySyncNeeded() {
+ logger.trace("Has remote function registry: {}", remoteFunctionRegistry.hasRegistry());
return remoteFunctionRegistry.hasRegistry() &&
isRegistrySyncNeeded(remoteFunctionRegistry.getRegistryVersion(), localFunctionRegistry.getVersion());
}
/**
* Checks if local function registry should be synchronized with remote function registry.
- * If remote function registry version is -1, it means that remote function registry is unreachable
- * or is not configured thus we skip synchronization and return false.
- * In all other cases synchronization is needed if remote and local function registries versions do not match.
+ *
+ * <ul>If remote function registry version is {@link DataChangeVersion#UNDEFINED},
+ * it means that remote function registry does not support versioning
+ * thus we need to synchronize both registries.</ul>
+ * <ul>If remote function registry version is {@link DataChangeVersion#NOT_AVAILABLE},
+ * it means that remote function registry is unreachable
+ * or is not configured thus we skip synchronization and return false.</ul>
+ * <ul>For all other cases synchronization is needed if remote
+ * and local function registries versions do not match.</ul>
*
* @param remoteVersion remote function registry version
* @param localVersion local function registry version
* @return true is local registry should be refreshed, false otherwise
*/
- private boolean isRegistrySyncNeeded(long remoteVersion, long localVersion) {
- return remoteVersion != -1 && remoteVersion != localVersion;
+ private boolean isRegistrySyncNeeded(int remoteVersion, int localVersion) {
+ logger.trace("Compare remote [{}] and local [{}] registry versions.", remoteVersion, localVersion);
+ return remoteVersion == DataChangeVersion.UNDEFINED ||
+ (remoteVersion != DataChangeVersion.NOT_AVAILABLE && remoteVersion != localVersion);
}
/**
@@ -459,7 +474,7 @@ public class FunctionImplementationRegistry implements FunctionLookupContext, Au
DataChangeVersion version) {
List<Jar> remoteJars = remoteFunctionRegistry.getRegistry(version).getJarList();
List<String> localJars = localFunctionRegistry.getAllJarNames();
- List<String> missingJars = Lists.newArrayList();
+ List<String> missingJars = new ArrayList<>();
for (Jar jar : remoteJars) {
if (!localJars.contains(jar.getName())) {
missingJars.add(jar.getName());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolder.java
index dc8fd74..d1d4fc9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolder.java
@@ -19,18 +19,18 @@ package org.apache.drill.exec.expr.fn.registry;
import org.apache.drill.shaded.guava.com.google.common.collect.ArrayListMultimap;
import org.apache.drill.shaded.guava.com.google.common.collect.ListMultimap;
-import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
-import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
-import org.apache.drill.shaded.guava.com.google.common.collect.Queues;
import org.apache.drill.common.AutoCloseables.Closeable;
import org.apache.drill.common.concurrent.AutoCloseableLock;
import org.apache.drill.exec.expr.fn.DrillFuncHolder;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -45,9 +45,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
* Holder is designed to allow concurrent reads and single writes to keep data consistent.
* This is achieved by {@link ReadWriteLock} implementation usage.
* Holder has number version which indicates remote function registry version number it is in sync with.
- *
+ * <p/>
* Structure example:
*
+ * <pre>
* JARS
* built-in -> upper -> upper(VARCHAR-REQUIRED)
* -> lower -> lower(VARCHAR-REQUIRED)
@@ -72,12 +73,12 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
*
* custom_lower -> custom_lower(VARCHAR-REQUIRED) -> function holder for custom_lower(VARCHAR-REQUIRED)
* -> custom_lower(VARCHAR-OPTIONAL) -> function holder for custom_lower(VARCHAR-OPTIONAL)
- *
+ * </pre>
* where
- * First.jar is jar name represented by String
- * upper is function name represented by String
- * upper(VARCHAR-REQUIRED) is signature name represented by String which consist of function name, list of input parameters
- * function holder for upper(VARCHAR-REQUIRED) is {@link DrillFuncHolder} initiated for each function.
+ * <li><b>First.jar</b> is jar name represented by {@link String}.</li>
+ * <li><b>upper</b> is function name represented by {@link String}.</li>
+ * <li><b>upper(VARCHAR-REQUIRED)</b> is signature name represented by String which consist of function name, list of input parameters.</li>
+ * <li><b>function holder for upper(VARCHAR-REQUIRED)</b> is {@link DrillFuncHolder} initiated for each function.</li>
*
*/
public class FunctionRegistryHolder {
@@ -88,7 +89,7 @@ public class FunctionRegistryHolder {
private final AutoCloseableLock readLock = new AutoCloseableLock(readWriteLock.readLock());
private final AutoCloseableLock writeLock = new AutoCloseableLock(readWriteLock.writeLock());
// remote function registry number, it is in sync with
- private long version;
+ private int version;
// jar name, Map<function name, Queue<function signature>
private final Map<String, Map<String, Queue<String>>> jars;
@@ -97,15 +98,15 @@ public class FunctionRegistryHolder {
private final Map<String, Map<String, DrillFuncHolder>> functions;
public FunctionRegistryHolder() {
- this.functions = Maps.newConcurrentMap();
- this.jars = Maps.newConcurrentMap();
+ this.functions = new ConcurrentHashMap<>();
+ this.jars = new ConcurrentHashMap<>();
}
/**
* This is read operation, so several users at a time can get this data.
* @return local function registry version number
*/
- public long getVersion() {
+ public int getVersion() {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()) {
return version;
}
@@ -122,12 +123,12 @@ public class FunctionRegistryHolder {
*
* @param newJars jars and list of their function holders, each contains function name, signature and holder
*/
- public void addJars(Map<String, List<FunctionHolder>> newJars, long version) {
+ public void addJars(Map<String, List<FunctionHolder>> newJars, int version) {
try (@SuppressWarnings("unused") Closeable lock = writeLock.open()) {
for (Map.Entry<String, List<FunctionHolder>> newJar : newJars.entrySet()) {
String jarName = newJar.getKey();
removeAllByJar(jarName);
- Map<String, Queue<String>> jar = Maps.newConcurrentMap();
+ Map<String, Queue<String>> jar = new ConcurrentHashMap<>();
jars.put(jarName, jar);
addFunctions(jar, newJar.getValue());
}
@@ -156,7 +157,7 @@ public class FunctionRegistryHolder {
*/
public List<String> getAllJarNames() {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()) {
- return Lists.newArrayList(jars.keySet());
+ return new ArrayList<>(jars.keySet());
}
}
@@ -171,7 +172,7 @@ public class FunctionRegistryHolder {
public List<String> getFunctionNamesByJar(String jarName) {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()){
Map<String, Queue<String>> functions = jars.get(jarName);
- return functions == null ? Lists.<String>newArrayList() : Lists.newArrayList(functions.keySet());
+ return functions == null ? new ArrayList<>() : new ArrayList<>(functions.keySet());
}
}
@@ -185,14 +186,14 @@ public class FunctionRegistryHolder {
* @param version version holder
* @return all functions which their holders
*/
- public ListMultimap<String, DrillFuncHolder> getAllFunctionsWithHolders(AtomicLong version) {
+ public ListMultimap<String, DrillFuncHolder> getAllFunctionsWithHolders(AtomicInteger version) {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()) {
if (version != null) {
version.set(this.version);
}
ListMultimap<String, DrillFuncHolder> functionsWithHolders = ArrayListMultimap.create();
for (Map.Entry<String, Map<String, DrillFuncHolder>> function : functions.entrySet()) {
- functionsWithHolders.putAll(function.getKey(), Lists.newArrayList(function.getValue().values()));
+ functionsWithHolders.putAll(function.getKey(), new ArrayList<>(function.getValue().values()));
}
return functionsWithHolders;
}
@@ -220,7 +221,7 @@ public class FunctionRegistryHolder {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()) {
ListMultimap<String, String> functionsWithSignatures = ArrayListMultimap.create();
for (Map.Entry<String, Map<String, DrillFuncHolder>> function : functions.entrySet()) {
- functionsWithSignatures.putAll(function.getKey(), Lists.newArrayList(function.getValue().keySet()));
+ functionsWithSignatures.putAll(function.getKey(), new ArrayList<>(function.getValue().keySet()));
}
return functionsWithSignatures;
}
@@ -236,13 +237,13 @@ public class FunctionRegistryHolder {
* @param version version holder
* @return list of function holders
*/
- public List<DrillFuncHolder> getHoldersByFunctionName(String functionName, AtomicLong version) {
+ public List<DrillFuncHolder> getHoldersByFunctionName(String functionName, AtomicInteger version) {
try (@SuppressWarnings("unused") Closeable lock = readLock.open()) {
if (version != null) {
version.set(this.version);
}
Map<String, DrillFuncHolder> holders = functions.get(functionName);
- return holders == null ? Lists.<DrillFuncHolder>newArrayList() : Lists.newArrayList(holders.values());
+ return holders == null ? new ArrayList<>() : new ArrayList<>(holders.values());
}
}
@@ -316,17 +317,13 @@ public class FunctionRegistryHolder {
final String functionName = function.getName();
Queue<String> jarFunctions = jar.get(functionName);
if (jarFunctions == null) {
- jarFunctions = Queues.newConcurrentLinkedQueue();
+ jarFunctions = new ConcurrentLinkedQueue<>();
jar.put(functionName, jarFunctions);
}
final String functionSignature = function.getSignature();
jarFunctions.add(functionSignature);
- Map<String, DrillFuncHolder> signatures = functions.get(functionName);
- if (signatures == null) {
- signatures = Maps.newConcurrentMap();
- functions.put(functionName, signatures);
- }
+ Map<String, DrillFuncHolder> signatures = functions.computeIfAbsent(functionName, k -> new ConcurrentHashMap<>());
signatures.put(functionSignature, function.getHolder());
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/LocalFunctionRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/LocalFunctionRegistry.java
index 3740a6c..cefbd8c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/LocalFunctionRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/LocalFunctionRegistry.java
@@ -24,8 +24,9 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.drill.exec.store.sys.store.DataChangeVersion;
import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
import org.apache.drill.shaded.guava.com.google.common.collect.ListMultimap;
import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
@@ -77,13 +78,16 @@ public class LocalFunctionRegistry {
private final FunctionRegistryHolder registryHolder;
/**
- * Registers all functions present in Drill classpath on start-up. All functions will be marked as built-in.
- * Built-in functions are not allowed to be unregistered. Initially sync registry version will be set to 0.
+ * Registers all functions present in Drill classpath on start-up.
+ * All functions will be marked as built-in. Built-in functions are not allowed to be unregistered.
+ * Since local function registry version is based on remote function registry version,
+ * initially sync version will be set to {@link DataChangeVersion#UNDEFINED}
+ * to ensure that upon first check both registries would be synchronized.
*/
public LocalFunctionRegistry(ScanResult classpathScan) {
registryHolder = new FunctionRegistryHolder();
validate(BUILT_IN, classpathScan);
- register(Lists.newArrayList(new JarScan(BUILT_IN, classpathScan, this.getClass().getClassLoader())), 0);
+ register(Lists.newArrayList(new JarScan(BUILT_IN, classpathScan, this.getClass().getClassLoader())), DataChangeVersion.UNDEFINED);
if (logger.isTraceEnabled()) {
StringBuilder allFunctions = new StringBuilder();
for (DrillFuncHolder method: registryHolder.getAllFunctionsWithHolders().values()) {
@@ -96,7 +100,7 @@ public class LocalFunctionRegistry {
/**
* @return remote function registry version number with which local function registry is synced
*/
- public long getVersion() {
+ public int getVersion() {
return registryHolder.getVersion();
}
@@ -160,7 +164,7 @@ public class LocalFunctionRegistry {
* @param jars list of jars to be registered
* @param version remote function registry version number with which local function registry is synced
*/
- public void register(List<JarScan> jars, long version) {
+ public void register(List<JarScan> jars, int version) {
Map<String, List<FunctionHolder>> newJars = new HashMap<>();
for (JarScan jarScan : jars) {
FunctionConverter converter = new FunctionConverter();
@@ -219,7 +223,7 @@ public class LocalFunctionRegistry {
* @param name function name
* @return all function holders associated with the function name. Function name is case insensitive.
*/
- public List<DrillFuncHolder> getMethods(String name, AtomicLong version) {
+ public List<DrillFuncHolder> getMethods(String name, AtomicInteger version) {
return registryHolder.getHoldersByFunctionName(name.toLowerCase(), version);
}
@@ -238,7 +242,7 @@ public class LocalFunctionRegistry {
* @param operatorTable drill operator table
*/
public void register(DrillOperatorTable operatorTable) {
- AtomicLong versionHolder = new AtomicLong();
+ AtomicInteger versionHolder = new AtomicInteger();
final Map<String, Collection<DrillFuncHolder>> registeredFunctions =
registryHolder.getAllFunctionsWithHolders(versionHolder).asMap();
operatorTable.setFunctionRegistryVersion(versionHolder.get());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/RemoteFunctionRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/RemoteFunctionRegistry.java
index 4e94765..f727a93 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/RemoteFunctionRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/registry/RemoteFunctionRegistry.java
@@ -54,34 +54,36 @@ import static com.fasterxml.jackson.databind.SerializationFeature.INDENT_OUTPUT;
* Creates all remote registry areas at startup and validates them,
* during init establishes connections with three udf related stores.
* Provides tools to work with three udf related stores, gives access to remote registry areas.
- *
+ * <p/>
* There are three udf stores:
- * REGISTRY - persistent store, stores remote function registry {@link Registry} under udf path
+ *
+ * <li><b>REGISTRY</b> - persistent store, stores remote function registry {@link Registry} under udf path
* which contains information about all dynamically registered jars and their function signatures.
- * If connection is created for the first time, puts empty remote registry.
+ * If connection is created for the first time, puts empty remote registry.</li>
*
- * UNREGISTRATION - transient store, stores information under udf/unregister path.
+ * <li><b>UNREGISTRATION</b> - transient store, stores information under udf/unregister path.
* udf/unregister path is persistent by itself but any child created will be transient.
* Whenever user submits request to unregister jar, child path with jar name is created under this store.
* This store also holds unregistration listener, which notifies all drill bits when child path is created,
- * so they can start local unregistration process.
+ * so they can start local unregistration process.</li>
*
- * JARS - transient store, stores information under udf/jars path.
+ * <li><b>JARS</b> - transient store, stores information under udf/jars path.
* udf/jars path is persistent by itself but any child created will be transient.
* Servers as lock, not allowing to perform any action on the same time.
* There two types of actions: {@link Action#REGISTRATION} and {@link Action#UNREGISTRATION}.
* Before starting any action, users tries to create child path with jar name under this store
* and if such path already exists, receives action being performed on that very jar.
- * When user finishes its action, he deletes child path with jar name.
- *
+ * When user finishes its action, he deletes child path with jar name.</li>
+ * <p/>
* There are three udf areas:
- * STAGING - area where user copies binary and source jars before starting registration process.
- * REGISTRY - area where registered jars are stored.
- * TMP - area where source and binary jars are backed up in unique folder during registration process.
+ *
+ * <li><b>STAGING</b> - area where user copies binary and source jars before starting registration process.</li>
+ * <li><b>REGISTRY</b> - area where registered jars are stored.</li>
+ * <li><b>TMP</b> - area where source and binary jars are backed up in unique folder during registration process.</li>
*/
public class RemoteFunctionRegistry implements AutoCloseable {
- private static final String registry_path = "registry";
+ private static final String REGISTRY_PATH = "registry";
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RemoteFunctionRegistry.class);
private static final ObjectMapper mapper = new ObjectMapper().enable(INDENT_OUTPUT);
@@ -112,19 +114,19 @@ public class RemoteFunctionRegistry implements AutoCloseable {
*
* @return remote function registry version if any, -1 otherwise
*/
- public long getRegistryVersion() {
+ public int getRegistryVersion() {
DataChangeVersion version = new DataChangeVersion();
boolean contains = false;
try {
- contains = registry.contains(registry_path, version);
+ contains = registry.contains(REGISTRY_PATH, version);
} catch (Exception e) {
- logger.error("Problem during trying to access remote function registry [{}]", registry_path, e);
+ logger.error("Problem during trying to access remote function registry [{}]", REGISTRY_PATH, e);
}
if (contains) {
return version.getVersion();
} else {
- logger.error("Remote function registry [{}] is unreachable", registry_path);
- return -1;
+ logger.error("Remote function registry [{}] is unreachable", REGISTRY_PATH);
+ return DataChangeVersion.NOT_AVAILABLE;
}
}
@@ -137,11 +139,11 @@ public class RemoteFunctionRegistry implements AutoCloseable {
public boolean hasRegistry() { return registry != null; }
public Registry getRegistry(DataChangeVersion version) {
- return registry.get(registry_path, version);
+ return registry.get(REGISTRY_PATH, version);
}
public void updateRegistry(Registry registryContent, DataChangeVersion version) throws VersionMismatchException {
- registry.put(registry_path, registryContent, version);
+ registry.put(REGISTRY_PATH, registryContent, version);
}
public void submitForUnregistration(String jar) {
@@ -193,7 +195,8 @@ public class RemoteFunctionRegistry implements AutoCloseable {
.persist()
.build();
registry = storeProvider.getOrCreateVersionedStore(registrationConfig);
- registry.putIfAbsent(registry_path, Registry.getDefaultInstance());
+ logger.trace("Remote function registry type: {}.", registry.getClass());
+ registry.putIfAbsent(REGISTRY_PATH, Registry.getDefaultInstance());
} catch (StoreException e) {
throw new DrillRuntimeException("Failure while loading remote registry.", e);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillOperatorTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillOperatorTable.java
index e1e3309..eb79a5a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillOperatorTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillOperatorTable.java
@@ -54,7 +54,7 @@ public class DrillOperatorTable extends SqlStdOperatorTable {
private final ArrayListMultimap<String, SqlOperator> drillOperatorsWithInferenceMap = ArrayListMultimap.create();
// indicates remote function registry version based on which drill operator were loaded
// is used to define if we need to reload operator table in case remote function registry version has changed
- private long functionRegistryVersion;
+ private int functionRegistryVersion;
private final OptionManager systemOptionManager;
@@ -70,14 +70,14 @@ public class DrillOperatorTable extends SqlStdOperatorTable {
*
* @param version registry version
*/
- public void setFunctionRegistryVersion(long version) {
+ public void setFunctionRegistryVersion(int version) {
functionRegistryVersion = version;
}
/**
* @return function registry version based on which operator table was loaded
*/
- public long getFunctionRegistryVersion() {
+ public int getFunctionRegistryVersion() {
return functionRegistryVersion;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
index e3cd7e4..41faea9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java
@@ -21,7 +21,6 @@ import java.io.IOException;
import org.apache.calcite.sql.SqlDescribeSchema;
import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.tools.RelConversionException;
import org.apache.calcite.tools.ValidationException;
import org.apache.drill.common.exceptions.UserException;
@@ -58,8 +57,7 @@ public class DrillSqlWorker {
* @param sql sql query
* @return query physical plan
*/
- public static PhysicalPlan getPlan(QueryContext context, String sql) throws SqlParseException, ValidationException,
- ForemanSetupException {
+ public static PhysicalPlan getPlan(QueryContext context, String sql) throws ForemanSetupException {
return getPlan(context, sql, null);
}
@@ -76,15 +74,18 @@ public class DrillSqlWorker {
* @param textPlan text plan
* @return query physical plan
*/
- public static PhysicalPlan getPlan(QueryContext context, String sql, Pointer<String> textPlan)
- throws ForemanSetupException {
+ public static PhysicalPlan getPlan(QueryContext context, String sql, Pointer<String> textPlan) throws ForemanSetupException {
Pointer<String> textPlanCopy = textPlan == null ? null : new Pointer<>(textPlan.value);
try {
return getQueryPlan(context, sql, textPlan);
} catch (Exception e) {
+ logger.trace("There was an error during conversion into physical plan. " +
+ "Will sync remote and local function registries if needed and retry " +
+ "in case if issue was due to missing function implementation.");
if (context.getFunctionRegistry().syncWithRemoteRegistry(
context.getDrillOperatorTable().getFunctionRegistryVersion())) {
context.reloadDrillOperatorTable();
+ logger.trace("Local function registry was synchronized with remote. Trying to find function one more time.");
return getQueryPlan(context, sql, textPlanCopy);
}
throw e;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/PersistentStoreProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/PersistentStoreProvider.java
index 4311f48..ca26a24 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/PersistentStoreProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/PersistentStoreProvider.java
@@ -18,24 +18,37 @@
package org.apache.drill.exec.store.sys;
import org.apache.drill.exec.exception.StoreException;
-import org.apache.drill.exec.store.sys.store.VersionedDelegatingStore;
+import org.apache.drill.exec.store.sys.store.UndefinedVersionDelegatingStore;
/**
* A factory used to create {@link PersistentStore store} instances.
- *
*/
public interface PersistentStoreProvider extends AutoCloseable {
+
/**
* Gets or creates a {@link PersistentStore persistent store} for the given configuration.
*
* Note that implementors have liberty to cache previous {@link PersistentStore store} instances.
*
- * @param config store configuration
- * @param <V> store value type
+ * @param config store configuration
+ * @param <V> store value type
+ * @return persistent store instance
+ * @throws StoreException in case when unable to create store
*/
<V> PersistentStore<V> getOrCreateStore(PersistentStoreConfig<V> config) throws StoreException;
+
+ /**
+ * Override this method if store supports versioning and return versioning instance.
+ * By default, undefined version wrapper will be used.
+ *
+ * @param config store configuration
+ * @param <V> store value type
+ * @return versioned persistent store instance
+ * @throws StoreException in case when unable to create store
+ */
default <V> VersionedPersistentStore<V> getOrCreateVersionedStore(PersistentStoreConfig<V> config) throws StoreException {
- return new VersionedDelegatingStore<>(getOrCreateStore(config));
+ // for those stores that do not support versioning
+ return new UndefinedVersionDelegatingStore<>(getOrCreateStore(config));
}
/**
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/DataChangeVersion.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/DataChangeVersion.java
index d182de3..76e5610 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/DataChangeVersion.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/DataChangeVersion.java
@@ -17,9 +17,17 @@
*/
package org.apache.drill.exec.store.sys.store;
+/**
+ * Holder for store version. By default version is {@link DataChangeVersion#UNDEFINED}.
+ */
public class DataChangeVersion {
- private int version;
+ // is used when store in unreachable
+ public static final int NOT_AVAILABLE = -1;
+ // is used when store does not support versioning
+ public static final int UNDEFINED = -2;
+
+ private int version = UNDEFINED;
public void setVersion(int version) {
this.version = version;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/UndefinedVersionDelegatingStore.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/UndefinedVersionDelegatingStore.java
new file mode 100644
index 0000000..5873ec0
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/UndefinedVersionDelegatingStore.java
@@ -0,0 +1,82 @@
+/*
+ * 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.store.sys.store;
+
+import org.apache.drill.exec.store.sys.PersistentStore;
+import org.apache.drill.exec.store.sys.PersistentStoreMode;
+import org.apache.drill.exec.store.sys.VersionedPersistentStore;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * Wrapper store that delegates operations to PersistentStore.
+ * Does not keep versioning and returns {@link DataChangeVersion#UNDEFINED} when version is required.
+ *
+ * @param <V> store value type
+ */
+public class UndefinedVersionDelegatingStore<V> implements VersionedPersistentStore<V> {
+
+ private final PersistentStore<V> store;
+
+ public UndefinedVersionDelegatingStore(PersistentStore<V> store) {
+ this.store = store;
+ }
+
+ @Override
+ public boolean contains(String key, DataChangeVersion version) {
+ version.setVersion(DataChangeVersion.UNDEFINED);
+ return store.contains(key);
+ }
+
+ @Override
+ public V get(String key, DataChangeVersion version) {
+ version.setVersion(DataChangeVersion.UNDEFINED);
+ return store.get(key);
+ }
+
+ @Override
+ public void put(String key, V value, DataChangeVersion version) {
+ store.put(key, value);
+ }
+
+ @Override
+ public PersistentStoreMode getMode() {
+ return store.getMode();
+ }
+
+ @Override
+ public void delete(String key) {
+ store.delete(key);
+ }
+
+ @Override
+ public boolean putIfAbsent(String key, V value) {
+ return store.putIfAbsent(key, value);
+ }
+
+ @Override
+ public Iterator<Map.Entry<String, V>> getRange(int skip, int take) {
+ return store.getRange(skip, take);
+ }
+
+ @Override
+ public void close() throws Exception {
+ store.close();
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/VersionedDelegatingStore.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/VersionedDelegatingStore.java
index 18e0b82..40576d5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/VersionedDelegatingStore.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/VersionedDelegatingStore.java
@@ -30,22 +30,24 @@ import org.apache.drill.exec.store.sys.PersistentStoreMode;
import org.apache.drill.exec.store.sys.VersionedPersistentStore;
/**
- * Versioned Store that delegates operations to PersistentStore
- * @param <V>
+ * Versioned store that delegates operations to PersistentStore and keeps versioning,
+ * incrementing version each time write / delete operation is triggered.
+ * Once created initial version is 0. Can be used only for local versioning, not distributed.
+ *
+ * @param <V> store value type
*/
public class VersionedDelegatingStore<V> implements VersionedPersistentStore<V> {
private final PersistentStore<V> store;
- private final ReadWriteLock readWriteLock;
private final AutoCloseableLock readLock;
private final AutoCloseableLock writeLock;
private int version;
public VersionedDelegatingStore(PersistentStore<V> store) {
this.store = store;
- readWriteLock = new ReentrantReadWriteLock();
+ ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
readLock = new AutoCloseableLock(readWriteLock.readLock());
writeLock = new AutoCloseableLock(readWriteLock.writeLock());
- version = -1;
+ version = 0;
}
@Override
@@ -113,7 +115,7 @@ public class VersionedDelegatingStore<V> implements VersionedPersistentStore<V>
{
try (@SuppressWarnings("unused") Closeable lock = writeLock.open()) {
store.close();
- version = -1;
+ version = DataChangeVersion.NOT_AVAILABLE;
}
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/CachingPersistentStoreProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/CachingPersistentStoreProvider.java
index aa6ee9d..75cef2f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/CachingPersistentStoreProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/CachingPersistentStoreProvider.java
@@ -17,21 +17,23 @@
*/
package org.apache.drill.exec.store.sys.store.provider;
+import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.ConcurrentMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
-import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
+import org.apache.drill.exec.store.sys.VersionedPersistentStore;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.exec.exception.StoreException;
import org.apache.drill.exec.store.sys.PersistentStore;
import org.apache.drill.exec.store.sys.PersistentStoreConfig;
import org.apache.drill.exec.store.sys.PersistentStoreProvider;
+import org.apache.drill.common.util.function.CheckedFunction;
public class CachingPersistentStoreProvider extends BasePersistentStoreProvider {
-// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CachingPersistentStoreProvider.class);
- private final ConcurrentMap<PersistentStoreConfig<?>, PersistentStore<?>> storeCache = Maps.newConcurrentMap();
+ private final Map<PersistentStoreConfig<?>, PersistentStore<?>> storeCache = new ConcurrentHashMap<>();
+ private final Map<PersistentStoreConfig<?>, VersionedPersistentStore<?>> versionedStoreCache = new ConcurrentHashMap<>();
private final PersistentStoreProvider provider;
public CachingPersistentStoreProvider(PersistentStoreProvider provider) {
@@ -41,21 +43,15 @@ public class CachingPersistentStoreProvider extends BasePersistentStoreProvider
@Override
@SuppressWarnings("unchecked")
public <V> PersistentStore<V> getOrCreateStore(final PersistentStoreConfig<V> config) throws StoreException {
- final PersistentStore<?> store = storeCache.get(config);
- if (store == null) {
- final PersistentStore<?> newStore = provider.getOrCreateStore(config);
- final PersistentStore<?> finalStore = storeCache.putIfAbsent(config, newStore);
- if (finalStore == null) {
- return (PersistentStore<V>)newStore;
- }
- try {
- newStore.close();
- } catch (Exception ex) {
- throw new StoreException(ex);
- }
- }
+ CheckedFunction<PersistentStoreConfig<?>, PersistentStore<?>, StoreException> function = provider::getOrCreateStore;
+ return (PersistentStore<V>) storeCache.computeIfAbsent(config, function);
+ }
- return (PersistentStore<V>) store;
+ @Override
+ @SuppressWarnings("unchecked")
+ public <V> VersionedPersistentStore<V> getOrCreateVersionedStore(PersistentStoreConfig<V> config) throws StoreException {
+ CheckedFunction<PersistentStoreConfig<?>, VersionedPersistentStore<?>, StoreException> function = provider::getOrCreateVersionedStore;
+ return (VersionedPersistentStore<V>) versionedStoreCache.computeIfAbsent(config, function);
}
@Override
@@ -65,12 +61,19 @@ public class CachingPersistentStoreProvider extends BasePersistentStoreProvider
@Override
public void close() throws Exception {
- final List<AutoCloseable> closeables = Lists.newArrayList();
- for (final AutoCloseable store : storeCache.values()) {
- closeables.add(store);
- }
- closeables.add(provider);
+ List<AutoCloseable> closeables = new ArrayList<>();
+
+ // add un-versioned stores
+ closeables.addAll(storeCache.values());
storeCache.clear();
+
+ // add versioned stores
+ closeables.addAll(versionedStoreCache.values());
+ versionedStoreCache.clear();
+
+ // add provider
+ closeables.add(provider);
+
AutoCloseables.close(closeables);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/InMemoryStoreProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/InMemoryStoreProvider.java
index 3ab85ec..6a70df7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/InMemoryStoreProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/InMemoryStoreProvider.java
@@ -17,11 +17,12 @@
*/
package org.apache.drill.exec.store.sys.store.provider;
-import org.apache.drill.exec.exception.StoreException;
import org.apache.drill.exec.store.sys.PersistentStore;
import org.apache.drill.exec.store.sys.PersistentStoreConfig;
import org.apache.drill.exec.store.sys.PersistentStoreProvider;
+import org.apache.drill.exec.store.sys.VersionedPersistentStore;
import org.apache.drill.exec.store.sys.store.InMemoryStore;
+import org.apache.drill.exec.store.sys.store.VersionedDelegatingStore;
public class InMemoryStoreProvider implements PersistentStoreProvider {
@@ -35,10 +36,15 @@ public class InMemoryStoreProvider implements PersistentStoreProvider {
public void close() throws Exception { }
@Override
- public <V> PersistentStore<V> getOrCreateStore(PersistentStoreConfig<V> config) throws StoreException {
+ public <V> PersistentStore<V> getOrCreateStore(PersistentStoreConfig<V> config) {
return new InMemoryStore<>(capacity);
}
@Override
- public void start() throws Exception { }
+ public <V> VersionedPersistentStore<V> getOrCreateVersionedStore(PersistentStoreConfig<V> config) {
+ return new VersionedDelegatingStore<>(getOrCreateStore(config));
+ }
+
+ @Override
+ public void start() { }
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/LocalPersistentStoreProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/LocalPersistentStoreProvider.java
index af67771..2dae62d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/LocalPersistentStoreProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/store/provider/LocalPersistentStoreProvider.java
@@ -26,7 +26,9 @@ import org.apache.drill.exec.store.dfs.DrillFileSystem;
import org.apache.drill.exec.store.sys.PersistentStore;
import org.apache.drill.exec.store.sys.PersistentStoreConfig;
import org.apache.drill.exec.store.sys.PersistentStoreRegistry;
+import org.apache.drill.exec.store.sys.VersionedPersistentStore;
import org.apache.drill.exec.store.sys.store.LocalPersistentStore;
+import org.apache.drill.exec.store.sys.store.VersionedDelegatingStore;
import org.apache.drill.exec.testing.store.NoWriteLocalStore;
import org.apache.hadoop.fs.Path;
@@ -70,6 +72,10 @@ public class LocalPersistentStoreProvider extends BasePersistentStoreProvider {
}
}
+ @Override
+ public <V> VersionedPersistentStore<V> getOrCreateVersionedStore(PersistentStoreConfig<V> config) {
+ return new VersionedDelegatingStore<>(getOrCreateStore(config));
+ }
@Override
public void close() throws Exception {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolderTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolderTest.java
index 8f5252d..9782315 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolderTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/expr/fn/registry/FunctionRegistryHolderTest.java
@@ -20,7 +20,6 @@ package org.apache.drill.exec.expr.fn.registry;
import org.apache.drill.shaded.guava.com.google.common.collect.ArrayListMultimap;
import org.apache.drill.shaded.guava.com.google.common.collect.ListMultimap;
import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
-import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
import org.apache.drill.categories.SqlFunctionTest;
import org.apache.drill.exec.expr.fn.DrillFuncHolder;
import org.junit.Before;
@@ -30,9 +29,11 @@ import org.junit.experimental.categories.Category;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -51,7 +52,7 @@ public class FunctionRegistryHolderTest {
@BeforeClass
public static void init() {
- newJars = Maps.newHashMap();
+ newJars = new HashMap<>();
FunctionHolder lower = new FunctionHolder("lower", "lower(VARCHAR-REQUIRED)", mock(DrillFuncHolder.class));
FunctionHolder upper = new FunctionHolder("upper", "upper(VARCHAR-REQUIRED)", mock(DrillFuncHolder.class));
newJars.put(built_in, Lists.newArrayList(lower, upper));
@@ -69,9 +70,9 @@ public class FunctionRegistryHolderTest {
@Test
public void testVersion() {
resetRegistry();
- long expectedVersion = 0;
+ int expectedVersion = 0;
assertEquals("Initial version should be 0", expectedVersion, registryHolder.getVersion());
- registryHolder.addJars(Maps.<String, List<FunctionHolder>>newHashMap(), ++expectedVersion);
+ registryHolder.addJars(new HashMap<>(), ++expectedVersion);
assertEquals("Version can change if no jars were added.", expectedVersion, registryHolder.getVersion());
fillInRegistry(++expectedVersion);
assertEquals("Version should have incremented by 1", expectedVersion, registryHolder.getVersion());
@@ -87,7 +88,7 @@ public class FunctionRegistryHolderTest {
public void testAddJars() {
resetRegistry();
int functionsSize = 0;
- List<String> jars = Lists.newArrayList();
+ List<String> jars = new ArrayList<>();
ListMultimap<String, DrillFuncHolder> functionsWithHolders = ArrayListMultimap.create();
ListMultimap<String, String> functionsWithSignatures = ArrayListMultimap.create();
for (Map.Entry<String, List<FunctionHolder>> jar : newJars.entrySet()) {
@@ -99,7 +100,7 @@ public class FunctionRegistryHolderTest {
}
}
- long expectedVersion = 0;
+ int expectedVersion = 0;
registryHolder.addJars(newJars, ++expectedVersion);
assertEquals("Version number should match", expectedVersion, registryHolder.getVersion());
compareTwoLists(jars, registryHolder.getAllJarNames());
@@ -112,7 +113,7 @@ public class FunctionRegistryHolderTest {
public void testAddTheSameJars() {
resetRegistry();
int functionsSize = 0;
- List<String> jars = Lists.newArrayList();
+ List<String> jars = new ArrayList<>();
ListMultimap<String, DrillFuncHolder> functionsWithHolders = ArrayListMultimap.create();
ListMultimap<String, String> functionsWithSignatures = ArrayListMultimap.create();
for (Map.Entry<String, List<FunctionHolder>> jar : newJars.entrySet()) {
@@ -123,7 +124,7 @@ public class FunctionRegistryHolderTest {
functionsSize++;
}
}
- long expectedVersion = 0;
+ int expectedVersion = 0;
registryHolder.addJars(newJars, ++expectedVersion);
assertEquals("Version number should match", expectedVersion, registryHolder.getVersion());
compareTwoLists(jars, registryHolder.getAllJarNames());
@@ -150,16 +151,15 @@ public class FunctionRegistryHolderTest {
@Test
public void testGetAllJarNames() {
- ArrayList<String> expectedResult = Lists.newArrayList(newJars.keySet());
+ List<String> expectedResult = new ArrayList<>(newJars.keySet());
compareTwoLists(expectedResult, registryHolder.getAllJarNames());
}
@Test
public void testGetFunctionNamesByJar() {
- ArrayList<String> expectedResult = Lists.newArrayList();
- for (FunctionHolder functionHolder : newJars.get(built_in)) {
- expectedResult.add(functionHolder.getName());
- }
+ List<String> expectedResult = newJars.get(built_in).stream()
+ .map(FunctionHolder::getName)
+ .collect(Collectors.toList());
compareTwoLists(expectedResult, registryHolder.getFunctionNamesByJar(built_in));
}
@@ -171,7 +171,7 @@ public class FunctionRegistryHolderTest {
expectedResult.put(functionHolder.getName(), functionHolder.getHolder());
}
}
- AtomicLong version = new AtomicLong();
+ AtomicInteger version = new AtomicInteger();
compareListMultimaps(expectedResult, registryHolder.getAllFunctionsWithHolders(version));
assertEquals("Version number should match", version.get(), registryHolder.getVersion());
}
@@ -200,30 +200,26 @@ public class FunctionRegistryHolderTest {
@Test
public void testGetHoldersByFunctionNameWithVersion() {
- List<DrillFuncHolder> expectedResult = Lists.newArrayList();
- for (List<FunctionHolder> functionHolders : newJars.values()) {
- for (FunctionHolder functionHolder : functionHolders) {
- if ("lower".equals(functionHolder.getName())) {
- expectedResult.add(functionHolder.getHolder());
- }
- }
- }
+ List<DrillFuncHolder> expectedResult = newJars.values().stream()
+ .flatMap(Collection::stream)
+ .filter(f -> "lower".equals(f.getName()))
+ .map(FunctionHolder::getHolder)
+ .collect(Collectors.toList());
+
assertFalse(expectedResult.isEmpty());
- AtomicLong version = new AtomicLong();
+ AtomicInteger version = new AtomicInteger();
compareTwoLists(expectedResult, registryHolder.getHoldersByFunctionName("lower", version));
assertEquals("Version number should match", version.get(), registryHolder.getVersion());
}
@Test
public void testGetHoldersByFunctionName() {
- List<DrillFuncHolder> expectedResult = Lists.newArrayList();
- for (List<FunctionHolder> functionHolders : newJars.values()) {
- for (FunctionHolder functionHolder : functionHolders) {
- if ("lower".equals(functionHolder.getName())) {
- expectedResult.add(functionHolder.getHolder());
- }
- }
- }
+ List<DrillFuncHolder> expectedResult = newJars.values().stream()
+ .flatMap(Collection::stream)
+ .filter(f -> "lower".equals(f.getName()))
+ .map(FunctionHolder::getHolder)
+ .collect(Collectors.toList());
+
assertFalse(expectedResult.isEmpty());
compareTwoLists(expectedResult, registryHolder.getHoldersByFunctionName("lower"));
}
@@ -236,10 +232,9 @@ public class FunctionRegistryHolderTest {
@Test
public void testFunctionsSize() {
- int count = 0;
- for (List<FunctionHolder> functionHolders : newJars.values()) {
- count += functionHolders.size();
- }
+ int count = newJars.values().stream()
+ .mapToInt(List::size)
+ .sum();
assertEquals("Functions size should match", count, registryHolder.functionsSize());
}
@@ -256,7 +251,7 @@ public class FunctionRegistryHolderTest {
registryHolder = new FunctionRegistryHolder();
}
- private void fillInRegistry(long version) {
+ private void fillInRegistry(int version) {
registryHolder.addJars(newJars, version);
}
@@ -266,7 +261,7 @@ public class FunctionRegistryHolderTest {
assertEquals("Multimaps size should match", m1.size(), m2.size());
for (Map.Entry<String, Collection<T>> entry : m1.entrySet()) {
try {
- compareTwoLists(Lists.newArrayList(entry.getValue()), Lists.newArrayList(m2.get(entry.getKey())));
+ compareTwoLists(new ArrayList<>(entry.getValue()), new ArrayList<>(m2.get(entry.getKey())));
} catch (AssertionError e) {
throw new AssertionError("Multimaps values should match", e);
}
@@ -275,9 +270,7 @@ public class FunctionRegistryHolderTest {
private <T> void compareTwoLists(List<T> l1, List<T> l2) {
assertEquals("Lists size should match", l1.size(), l2.size());
- for (T item : l1) {
- assertTrue("Two lists should have the same values", l2.contains(item));
- }
+ l1.forEach(i -> assertTrue("Two lists should have the same values", l2.contains(i)));
}
}
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/udf/dynamic/TestDynamicUDFSupport.java b/exec/java-exec/src/test/java/org/apache/drill/exec/udf/dynamic/TestDynamicUDFSupport.java
index a5a3c51..16ac42e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/udf/dynamic/TestDynamicUDFSupport.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/udf/dynamic/TestDynamicUDFSupport.java
@@ -64,7 +64,7 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
@@ -762,7 +762,7 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
DataChangeVersion version = new DataChangeVersion();
Registry registry = remoteFunctionRegistry.getRegistry(version);
- assertEquals("Remote registry version should match", 1, version.getVersion());
+ assertEquals("Remote registry version should match", 2, version.getVersion());
List<Jar> jarList = registry.getJarList();
assertEquals("Only one jar should be registered", 1, jarList.size());
assertEquals("Jar name should match", jar1, jarList.get(0).getName());
@@ -823,7 +823,7 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
DataChangeVersion version = new DataChangeVersion();
Registry registry = remoteFunctionRegistry.getRegistry(version);
- assertEquals("Remote registry version should match", 2, version.getVersion());
+ assertEquals("Remote registry version should match", 3, version.getVersion());
List<Jar> actualJars = registry.getJarList();
List<String> expectedJars = Lists.newArrayList(jar1, jar2);
@@ -861,7 +861,7 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
assertTrue("syncWithRemoteRegistry() should return true", result);
return true;
})
- .when(functionImplementationRegistry).syncWithRemoteRegistry(anyLong());
+ .when(functionImplementationRegistry).syncWithRemoteRegistry(anyInt());
SimpleQueryRunner simpleQueryRunner = new SimpleQueryRunner(query);
Thread thread1 = new Thread(simpleQueryRunner);
@@ -873,10 +873,10 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
thread1.join();
thread2.join();
- verify(functionImplementationRegistry, times(2)).syncWithRemoteRegistry(anyLong());
+ verify(functionImplementationRegistry, times(2)).syncWithRemoteRegistry(anyInt());
LocalFunctionRegistry localFunctionRegistry = (LocalFunctionRegistry)FieldUtils.readField(
functionImplementationRegistry, "localFunctionRegistry", true);
- assertEquals("Sync function registry version should match", 1L, localFunctionRegistry.getVersion());
+ assertEquals("Sync function registry version should match", 2, localFunctionRegistry.getVersion());
}
@Test
@@ -895,7 +895,7 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
assertFalse("syncWithRemoteRegistry() should return false", result);
return false;
})
- .when(functionImplementationRegistry).syncWithRemoteRegistry(anyLong());
+ .when(functionImplementationRegistry).syncWithRemoteRegistry(anyInt());
test("select custom_lower('A') from (values(1))");
@@ -906,10 +906,10 @@ public class TestDynamicUDFSupport extends BaseTestQuery {
assertThat(e.getMessage(), containsString("No match found for function signature unknown_lower(<CHARACTER>)"));
}
- verify(functionImplementationRegistry, times(2)).syncWithRemoteRegistry(anyLong());
+ verify(functionImplementationRegistry, times(2)).syncWithRemoteRegistry(anyInt());
LocalFunctionRegistry localFunctionRegistry = (LocalFunctionRegistry)FieldUtils.readField(
functionImplementationRegistry, "localFunctionRegistry", true);
- assertEquals("Sync function registry version should match", 1L, localFunctionRegistry.getVersion());
+ assertEquals("Sync function registry version should match", 2, localFunctionRegistry.getVersion());
}
private static String buildJars(String jarName, String includeFiles, String includeResources) {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/PrintingUtils.java b/exec/java-exec/src/test/java/org/apache/drill/test/PrintingUtils.java
index 1709bdf..c71a2d4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/PrintingUtils.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/PrintingUtils.java
@@ -19,7 +19,7 @@ package org.apache.drill.test;
import ch.qos.logback.classic.Level;
import org.apache.drill.exec.client.LoggingResultsListener;
-import org.apache.drill.exec.util.CheckedSupplier;
+import org.apache.drill.common.util.function.CheckedSupplier;
import org.apache.drill.exec.util.VectorUtil;
import java.util.function.Supplier;
[drill] 05/06: DRILL-6773: The renamed schema with aliases is not
shown for queries on empty directories
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit d5bafeccfc4d8dc19b7f22cfae807fa56b62340a
Author: Vitalii Diravka <vi...@gmail.com>
AuthorDate: Fri Oct 5 16:21:55 2018 +0300
DRILL-6773: The renamed schema with aliases is not shown for queries on empty directories
closes #1492
---
.../store/mapr/db/MapRDBPushFilterIntoScan.java | 4 +--
.../exec/store/hbase/HBasePushFilterIntoScan.java | 2 +-
.../store/kafka/KafkaPushDownFilterIntoScan.java | 2 +-
.../store/mongo/MongoPushDownFilterForScan.java | 2 +-
.../drill/exec/physical/base/SchemalessScan.java | 6 ++++
.../physical/impl/project/ProjectRecordBatch.java | 2 +-
.../planner/physical/ConvertCountToDirectScan.java | 2 +-
.../exec/planner/physical/ProjectAllowDupPrel.java | 14 +++++---
.../drill/exec/planner/physical/ProjectPrel.java | 2 +-
.../drill/exec/planner/physical/ScanPrel.java | 12 +++----
.../drill/exec/planner/physical/ScanPrule.java | 2 +-
.../physical/visitor/StarColumnConverter.java | 26 ++++++++++-----
.../planner/sql/handlers/DefaultSqlHandler.java | 1 +
.../InfoSchemaPushFilterIntoRecordGenerator.java | 2 +-
.../exec/store/parquet/ParquetPushDownFilter.java | 2 +-
.../test/java/org/apache/drill/TestUnionAll.java | 4 ++-
.../java/org/apache/drill/TestUnionDistinct.java | 4 ++-
.../org/apache/drill/exec/TestEmptyInputSql.java | 37 +++++++++++++++++++---
18 files changed, 89 insertions(+), 37 deletions(-)
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
index 0690531..c233a6b 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBPushFilterIntoScan.java
@@ -141,7 +141,7 @@ public abstract class MapRDBPushFilterIntoScan extends StoragePluginOptimizerRul
final JsonTableGroupScan newGroupsScan = groupScan.clone(newScanSpec);
newGroupsScan.setFilterPushedDown(true);
- final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+ final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
// Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of((RelNode)newScanPrel));
@@ -187,7 +187,7 @@ public abstract class MapRDBPushFilterIntoScan extends StoragePluginOptimizerRul
groupScan.getTableStats());
newGroupsScan.setFilterPushedDown(true);
- final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+ final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
// Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of((RelNode)newScanPrel));;
diff --git a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
index 692268f..736b36b 100644
--- a/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
+++ b/contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBasePushFilterIntoScan.java
@@ -122,7 +122,7 @@ public abstract class HBasePushFilterIntoScan extends StoragePluginOptimizerRule
newScanSpec, groupScan.getColumns());
newGroupsScan.setFilterPushedDown(true);
- final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+ final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
// Depending on whether is a project in the middle, assign either scan or copy of project to childRel.
final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of(newScanPrel));
diff --git a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
index 0e04f0f..14abadf 100644
--- a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
+++ b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaPushDownFilterIntoScan.java
@@ -66,7 +66,7 @@ public class KafkaPushDownFilterIntoScan extends StoragePluginOptimizerRule {
logger.info("Partitions ScanSpec after pushdown: " + newScanSpec);
GroupScan newGroupScan = groupScan.cloneWithNewSpec(newScanSpec);
- final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(), newGroupScan, scan.getRowType());
+ final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(), newGroupScan, scan.getRowType());
call.transformTo(filter.copy(filter.getTraitSet(), ImmutableList.of(newScanPrel)));
}
diff --git a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
index 540af07..afe32f2 100644
--- a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
+++ b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoPushDownFilterForScan.java
@@ -77,7 +77,7 @@ public class MongoPushDownFilterForScan extends StoragePluginOptimizerRule {
}
newGroupsScan.setFilterPushedDown(true);
- final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(),
+ final ScanPrel newScanPrel = new ScanPrel(scan, filter.getTraitSet(),
newGroupsScan, scan.getRowType());
if (mongoFilterBuilder.isAllExpressionsConverted()) {
/*
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/SchemalessScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/SchemalessScan.java
index 1db83f5..d227b04 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/SchemalessScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/SchemalessScan.java
@@ -92,6 +92,12 @@ public class SchemalessScan extends AbstractFileGroupScan implements SubScan {
return ScanStats.ZERO_RECORD_TABLE;
}
+
+ @Override
+ public boolean canPushdownProjects(List<SchemaPath> columns) {
+ return false;
+ }
+
@Override
public boolean supportsPartitionFilterPushdown() {
return false;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
index 8ea15d3..a051d99 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java
@@ -876,7 +876,7 @@ public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project> {
*/
@Override
protected IterOutcome handleNullInput() {
- if (! popConfig.isOutputProj()) {
+ if (!popConfig.isOutputProj()) {
return super.handleNullInput();
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ConvertCountToDirectScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ConvertCountToDirectScan.java
index 5ef97b9..f8cda8c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ConvertCountToDirectScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ConvertCountToDirectScan.java
@@ -128,7 +128,7 @@ public class ConvertCountToDirectScan extends Prule {
final ScanStats scanStats = new ScanStats(ScanStats.GroupScanProperty.EXACT_ROW_COUNT, 1, 1, scanRowType.getFieldCount());
final GroupScan directScan = new MetadataDirectGroupScan(reader, oldGrpScan.getFiles(), scanStats);
- final ScanPrel newScan = ScanPrel.create(scan,
+ final ScanPrel newScan = new ScanPrel(scan,
scan.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(DrillDistributionTrait.SINGLETON), directScan,
scanRowType);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectAllowDupPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectAllowDupPrel.java
index 55e6eaa..07936f3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectAllowDupPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectAllowDupPrel.java
@@ -39,13 +39,18 @@ import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
public class ProjectAllowDupPrel extends ProjectPrel {
public ProjectAllowDupPrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, List<RexNode> exps,
- RelDataType rowType) {
- super(cluster, traits, child, exps, rowType);
+ RelDataType rowType) {
+ this(cluster, traits, child, exps, rowType, false);
+ }
+
+ public ProjectAllowDupPrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, List<RexNode> exps,
+ RelDataType rowType, boolean outputProj) {
+ super(cluster, traits, child, exps, rowType, outputProj);
}
@Override
public ProjectAllowDupPrel copy(RelTraitSet traitSet, RelNode input, List<RexNode> exps, RelDataType rowType) {
- return new ProjectAllowDupPrel(getCluster(), traitSet, input, exps, rowType);
+ return new ProjectAllowDupPrel(getCluster(), traitSet, input, exps, rowType, outputProj);
}
@Override
@@ -54,7 +59,8 @@ public class ProjectAllowDupPrel extends ProjectPrel {
PhysicalOperator childPOP = child.getPhysicalOperator(creator);
- Project p = new Project(this.getProjectExpressions(new DrillParseContext(PrelUtil.getSettings(getCluster()))), childPOP);
+ Project p = new Project(this.getProjectExpressions(new DrillParseContext(PrelUtil.getSettings(getCluster()))),
+ childPOP, outputProj);
return creator.addMetadata(this, p);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
index 4f06b64..395e778 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ProjectPrel.java
@@ -48,7 +48,7 @@ import org.apache.calcite.sql.SqlKind;
public class ProjectPrel extends DrillProjectRelBase implements Prel{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProjectPrel.class);
- private final boolean outputProj;
+ protected final boolean outputProj;
public ProjectPrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, List<RexNode> exps,
RelDataType rowType) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
index a2655b3..ff901c9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
@@ -48,8 +48,11 @@ public class ScanPrel extends AbstractRelNode implements DrillScanPrel {
protected final GroupScan groupScan;
private final RelDataType rowType;
- public ScanPrel(RelOptCluster cluster, RelTraitSet traits,
- GroupScan groupScan, RelDataType rowType) {
+ public ScanPrel(RelNode old, RelTraitSet traitSets, GroupScan scan, RelDataType rowType) {
+ this(old.getCluster(), traitSets, scan, rowType);
+ }
+
+ public ScanPrel(RelOptCluster cluster, RelTraitSet traits, GroupScan groupScan, RelDataType rowType) {
super(cluster, traits);
this.groupScan = getCopy(groupScan);
this.rowType = rowType;
@@ -86,11 +89,6 @@ public class ScanPrel extends AbstractRelNode implements DrillScanPrel {
return groupScan;
}
- public static ScanPrel create(RelNode old, RelTraitSet traitSets,
- GroupScan scan, RelDataType rowType) {
- return new ScanPrel(old.getCluster(), traitSets, getCopy(scan), rowType);
- }
-
@Override
public RelWriter explainTerms(RelWriter pw) {
return super.explainTerms(pw).item("groupscan", groupScan.getDigest());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrule.java
index 4f66e90..57c4517 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrule.java
@@ -44,7 +44,7 @@ public class ScanPrule extends Prule{
final RelTraitSet traits = scan.getTraitSet().plus(Prel.DRILL_PHYSICAL).plus(partition);
- final DrillScanPrel newScan = ScanPrel.create(scan, traits, groupScan, scan.getRowType());
+ final DrillScanPrel newScan = new ScanPrel(scan, traits, groupScan, scan.getRowType());
call.transformTo(newScan);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
index ac5ec82..91eeffe 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
@@ -76,12 +76,12 @@ public class StarColumnConverter extends BasePrelVisitor<Prel, Void, RuntimeExce
Prel child = ((Prel) prel.getInput(0)).accept(this, null);
if (prefixedForStar) {
- if (!prefixedForWriter) {
+ if (prefixedForWriter) {
+ // Prefix is added under CTAS Writer. We need create a new Screen with the converted child.
+ return prel.copy(prel.getTraitSet(), Collections.singletonList(child));
+ } else {
// Prefix is added for SELECT only, not for CTAS writer.
return insertProjUnderScreenOrWriter(prel, prel.getInput().getRowType(), child);
- } else {
- // Prefix is added under CTAS Writer. We need create a new Screen with the converted child.
- return prel.copy(prel.getTraitSet(), Collections.<RelNode>singletonList(child));
}
} else {
// No prefix is
@@ -118,13 +118,23 @@ public class StarColumnConverter extends BasePrelVisitor<Prel, Void, RuntimeExce
RelDataType newRowType = RexUtil.createStructType(child.getCluster().getTypeFactory(),
exprs, origRowType.getFieldNames(), null);
- int fieldCount = prel.getRowType().isStruct()? prel.getRowType().getFieldCount():1;
+ int fieldCount = prel.getRowType().isStruct() ? prel.getRowType().getFieldCount() : 1;
// Insert PUS/PUW : remove the prefix and keep the original field name.
- if (fieldCount > 1) { // // no point in allowing duplicates if we only have one column
- proj = new ProjectAllowDupPrel(child.getCluster(), child.getTraitSet(), child, exprs, newRowType);
+ if (fieldCount > 1) { // no point in allowing duplicates if we only have one column
+ proj = new ProjectAllowDupPrel(child.getCluster(),
+ child.getTraitSet(),
+ child,
+ exprs,
+ newRowType,
+ true); //outputProj = true : will allow to build the schema for PUS Project, see ProjectRecordBatch#handleNullInput()
} else {
- proj = new ProjectPrel(child.getCluster(), child.getTraitSet(), child, exprs, newRowType);
+ proj = new ProjectPrel(child.getCluster(),
+ child.getTraitSet(),
+ child,
+ exprs,
+ newRowType,
+ true); //outputProj = true : will allow to build the schema for PUS Project, see ProjectRecordBatch#handleNullInput()
}
children.add(proj);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
index bad1617..c75311f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java
@@ -506,6 +506,7 @@ public class DefaultSqlHandler extends AbstractSqlHandler {
* The rest of projects will remove the duplicate column when we generate POP in json format.
*/
phyRelNode = StarColumnConverter.insertRenameProject(phyRelNode);
+ log("Physical RelNode after Top and Rename Project inserting: ", phyRelNode, logger, null);
/*
* 2.)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaPushFilterIntoRecordGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaPushFilterIntoRecordGenerator.java
index 3b62dae..dae5c28 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaPushFilterIntoRecordGenerator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaPushFilterIntoRecordGenerator.java
@@ -99,7 +99,7 @@ public abstract class InfoSchemaPushFilterIntoRecordGenerator extends StoragePlu
final InfoSchemaGroupScan newGroupsScan = new InfoSchemaGroupScan(groupScan.getTable(), infoSchemaFilter);
newGroupsScan.setFilterPushedDown(true);
- RelNode input = ScanPrel.create(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
+ RelNode input = new ScanPrel(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType());
if (project != null) {
input = project.copy(project.getTraitSet(), input, project.getProjects(), filter.getRowType());
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java
index 2057ac5..e58d306 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java
@@ -167,7 +167,7 @@ public abstract class ParquetPushDownFilter extends StoragePluginOptimizerRule {
return;
}
- RelNode newScan = ScanPrel.create(scan, scan.getTraitSet(), newGroupScan, scan.getRowType());
+ RelNode newScan = new ScanPrel(scan, scan.getTraitSet(), newGroupScan, scan.getRowType());
if (project != null) {
newScan = project.copy(project.getTraitSet(), ImmutableList.of(newScan));
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java b/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
index d1d5654..9185939 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestUnionAll.java
@@ -1248,7 +1248,9 @@ public class TestUnionAll extends BaseTestQuery {
@Test
public void testUnionAllBothEmptyDirs() throws Exception {
- final BatchSchema expectedSchema = new SchemaBuilder().build();
+ final BatchSchema expectedSchema = new SchemaBuilder()
+ .addNullable("key", TypeProtos.MinorType.INT)
+ .build();
testBuilder()
.sqlQuery("SELECT key FROM dfs.tmp.`%1$s` UNION ALL SELECT key FROM dfs.tmp.`%1$s`", EMPTY_DIR_NAME)
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestUnionDistinct.java b/exec/java-exec/src/test/java/org/apache/drill/TestUnionDistinct.java
index db43512..5910817 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestUnionDistinct.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestUnionDistinct.java
@@ -826,7 +826,9 @@ public class TestUnionDistinct extends BaseTestQuery {
@Test
public void testUnionBothEmptyDirs() throws Exception {
- final BatchSchema expectedSchema = new SchemaBuilder().build();
+ final BatchSchema expectedSchema = new SchemaBuilder()
+ .addNullable("key", TypeProtos.MinorType.INT)
+ .build();
testBuilder()
.sqlQuery("SELECT key FROM dfs.tmp.`%1$s` UNION SELECT key FROM dfs.tmp.`%1$s`", EMPTY_DIR_NAME)
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/TestEmptyInputSql.java b/exec/java-exec/src/test/java/org/apache/drill/exec/TestEmptyInputSql.java
index 1512059..2b0eb42 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/TestEmptyInputSql.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/TestEmptyInputSql.java
@@ -222,12 +222,9 @@ public class TestEmptyInputSql extends BaseTestQuery {
@Test
public void testEmptyDirectoryAndFieldInQuery() throws Exception {
- final List<Pair<SchemaPath, TypeProtos.MajorType>> expectedSchema = Lists.newArrayList();
- final TypeProtos.MajorType majorType = TypeProtos.MajorType.newBuilder()
- .setMinorType(TypeProtos.MinorType.INT) // field "key" is absent in schemaless table
- .setMode(TypeProtos.DataMode.OPTIONAL)
+ final BatchSchema expectedSchema = new SchemaBuilder()
+ .addNullable("key", TypeProtos.MinorType.INT)
.build();
- expectedSchema.add(Pair.of(SchemaPath.getSimplePath("key"), majorType));
testBuilder()
.sqlQuery("select key from dfs.tmp.`%s`", EMPTY_DIR_NAME)
@@ -237,6 +234,36 @@ public class TestEmptyInputSql extends BaseTestQuery {
}
@Test
+ public void testRenameProjectEmptyDirectory() throws Exception {
+ final BatchSchema expectedSchema = new SchemaBuilder()
+ .addNullable("WeekId", TypeProtos.MinorType.INT)
+ .addNullable("ProductName", TypeProtos.MinorType.INT)
+ .build();
+
+ testBuilder()
+ .sqlQuery("select WeekId, Product as ProductName from (select CAST(`dir0` as INT) AS WeekId, " +
+ "Product from dfs.tmp.`%s`)", EMPTY_DIR_NAME)
+ .schemaBaseLine(expectedSchema)
+ .build()
+ .run();
+ }
+
+ @Test
+ public void testRenameProjectEmptyJson() throws Exception {
+ final BatchSchema expectedSchema = new SchemaBuilder()
+ .addNullable("WeekId", TypeProtos.MinorType.INT)
+ .addNullable("ProductName", TypeProtos.MinorType.INT)
+ .build();
+
+ testBuilder()
+ .sqlQuery("select WeekId, Product as ProductName from (select CAST(`dir0` as INT) AS WeekId, " +
+ "Product from cp.`%s`)", SINGLE_EMPTY_JSON)
+ .schemaBaseLine(expectedSchema)
+ .build()
+ .run();
+ }
+
+ @Test
public void testEmptyDirectoryPlanSerDe() throws Exception {
String query = String.format("select * from dfs.tmp.`%s`", EMPTY_DIR_NAME);
PlanTestBase.testPhysicalPlanExecutionBasedOnQuery(query);
[drill] 06/06: DRILL-6541: Upgrade ZooKeeper patch version to
3.4.11 for mapr profile
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit a2533917b0bc387711060c739dfb9844896fe4be
Author: Bohdan Kazydub <bo...@gmail.com>
AuthorDate: Tue Jun 26 13:09:15 2018 +0300
DRILL-6541: Upgrade ZooKeeper patch version to 3.4.11 for mapr profile
closes #1398
---
contrib/storage-hbase/pom.xml | 1 +
pom.xml | 2 +-
2 files changed, 2 insertions(+), 1 deletion(-)
diff --git a/contrib/storage-hbase/pom.xml b/contrib/storage-hbase/pom.xml
index dcedc02..1423c26 100644
--- a/contrib/storage-hbase/pom.xml
+++ b/contrib/storage-hbase/pom.xml
@@ -263,6 +263,7 @@
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-testing-util</artifactId>
+ <scope>test</scope>
<classifier>tests</classifier>
<exclusions>
<exclusion>
diff --git a/pom.xml b/pom.xml
index 4d81927..5d00ee6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2442,7 +2442,7 @@
<hive.version>2.1.1-mapr-1710</hive.version>
<hbase.version>1.1.1-mapr-1602-m7-5.2.0</hbase.version>
<hadoop.version>2.7.0-mapr-1707</hadoop.version>
- <zookeeper.version>3.4.5-mapr-1710</zookeeper.version>
+ <zookeeper.version>3.4.11-mapr-1808</zookeeper.version>
</properties>
<dependencyManagement>
<dependencies>
[drill] 04/06: DRILL-6764: Query fails with IOB when Unnest has
reference to deep nested field like (t.c_orders.o_lineitems).
Posted by vo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 778e043f284921792403f6dc595050793a47ee02
Author: Hanumath Rao Maduri <ha...@gmail.com>
AuthorDate: Tue Oct 2 16:32:38 2018 -0700
DRILL-6764: Query fails with IOB when Unnest has reference to deep nested field like (t.c_orders.o_lineitems).
closes #1487
---
.../planner/sql/handlers/ComplexUnnestVisitor.java | 11 +-
.../impl/lateraljoin/TestLateralPlans.java | 20 +++
.../resources/lateraljoin/nested-customer-map.json | 134 +++++++++++++++++++++
3 files changed, 161 insertions(+), 4 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ComplexUnnestVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ComplexUnnestVisitor.java
index b8d62f5..60e46ef 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ComplexUnnestVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ComplexUnnestVisitor.java
@@ -82,15 +82,18 @@ public class ComplexUnnestVisitor extends RelShuttleImpl {
builder.push(newCorrelate);
List<RexNode> topProjectExpressions = left.getRowType().getFieldList().stream()
- .map(field -> builder.getRexBuilder().makeInputRef(left, field.getIndex()))
- .collect(Collectors.toList());
+ .map(field -> builder.getRexBuilder().makeInputRef(newCorrelate, field.getIndex()))
+ .collect(Collectors.toList());
+ //Accommodate the new $COMPLEX_FIELD_NAME column.
+ int rightStartIndex = left.getRowType().getFieldList().size() + 1;
switch (correlate.getJoinType()) {
case LEFT:
case INNER:
// adds field from the right input of correlate to the top project
- topProjectExpressions.add(
- builder.getRexBuilder().makeInputRef(newCorrelate, topProjectExpressions.size() + 1));
+ topProjectExpressions.addAll(right.getRowType().getFieldList().stream()
+ .map(field -> builder.getRexBuilder().makeInputRef(newCorrelate, field.getIndex() + rightStartIndex))
+ .collect(Collectors.toList()));
// fall through
case ANTI:
case SEMI:
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/lateraljoin/TestLateralPlans.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/lateraljoin/TestLateralPlans.java
index 9adc3b5..ea78ddd 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/lateraljoin/TestLateralPlans.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/lateraljoin/TestLateralPlans.java
@@ -549,4 +549,24 @@ public class TestLateralPlans extends BaseTestQuery {
.go();
}
}
+
+ @Test
+ public void testNestedColumnQuery() throws Exception {
+ String sql = "select dt.area_code as area_code, dt.ph as ph from cp.`lateraljoin/nested-customer-map.json` t," +
+ " lateral (select t2.ord.area_code as area_code , t2.ord.phone as ph from unnest(t.c_address.c_phone) t2(ord)) dt";
+
+ String baselineQuery = "select dt.c_ph.area_code as area_code, dt.c_ph.phone as ph from (select flatten(t.c_address.c_phone) as c_ph from cp.`lateraljoin/nested-customer-map.json` t) dt";
+
+ ClusterFixtureBuilder builder = ClusterFixture.builder(dirTestWatcher)
+ .setOptionDefault(PlannerSettings.ENABLE_UNNEST_LATERAL_KEY, true);
+
+ try (ClusterFixture cluster = builder.build();
+ ClientFixture client = cluster.clientFixture()) {
+ client.testBuilder()
+ .ordered()
+ .sqlBaselineQuery(baselineQuery)
+ .sqlQuery(sql)
+ .go();
+ }
+ }
}
diff --git a/exec/java-exec/src/test/resources/lateraljoin/nested-customer-map.json b/exec/java-exec/src/test/resources/lateraljoin/nested-customer-map.json
new file mode 100644
index 0000000..07de054
--- /dev/null
+++ b/exec/java-exec/src/test/resources/lateraljoin/nested-customer-map.json
@@ -0,0 +1,134 @@
+{
+ "c_name" : "customer1",
+ "c_id" : 1,
+ "c_phone" : ["6505200001", "4085201234", "6125205678"],
+ "orders" : [{"o_id": 1, "o_shop": "Meno Park 1st", "o_amount": 4.5,
+ "items" : [ {"i_name" : "paper towel", "i_number": 2, "i_supplier": "oregan"},
+ {"i_name" : "map", "i_number": 1, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 9, "i_supplier": "california"}
+ ]
+
+ },
+ {"o_id": 2, "o_shop": "Mountain View 1st", "o_amount": 104.5,
+ "items" : [ {"i_name" : "beef", "i_number": 3, "i_supplier": "montana"},
+ {"i_name" : "tooth paste", "i_number": 4, "i_supplier": "washington"},
+ {"i_name" : "hat", "i_number": 7, "i_supplier": "california"}
+ ]
+
+ },
+ {"o_id": 3, "o_shop": "Sunnyvale 1st", "o_amount": 294.5,
+ "items" : [ {"i_name" : "paper towel", "i_number": 5, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 6, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 8, "i_supplier": "california"}
+ ]
+ }
+ ],
+ "c_address" : {
+ "Area": "bay area",
+ "City": "Santa Clara",
+ "State" : "CA",
+ "c_phone": [{"area_code": "650", "phone": "5200001"}, {"area_code": "408", "phone": "5201234"}, {"area_code" : "612", "phone": "5205678"}]
+ }
+}
+{
+ "c_name" : "customer2",
+ "c_id" : 2,
+ "c_phone" : ["1505200001", "7085201234", "2125205678"],
+ "orders" : [{"o_id": 10, "o_shop": "Mountain View 1st", "o_amount": 724.5,
+ "items" : [ {"i_name" : "beef", "i_number": 12, "i_supplier": "montana"},
+ {"i_name" : "tooth paste", "i_number": 11, "i_supplier": "washington"},
+ {"i_name" : "hat", "i_number": 10, "i_supplier": "california"}
+ ]
+
+ },
+
+ {"o_id": 11, "o_shop": "Sunnyvale 1st", "o_amount": 179.5,
+ "items" : [ {"i_name" : "paper towel", "i_number": 13, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 14, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 15, "i_supplier": "california"}
+ ]
+ },
+ {"o_id": 12, "o_shop": "Meno Park 1st", "o_amount": 80.0,
+ "items" : [ {"i_name" : "paper towel", "i_number": 13, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 14, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 15, "i_supplier": "california"}
+ ]
+ }
+ ],
+ "c_address" : {
+ "Area": "Greater LA",
+ "City": "LA",
+ "State" : "CA",
+ "c_phone": [{"area_code": "150", "phone" : "5200001"}, {"area_code": "708", "phone": "5201234"}, {"area_code" : "212", "phone": "5205678"}]
+ }
+}
+{
+ "c_name" : "customer3",
+ "c_id" : 3,
+ "c_phone" : ["1205200001", "7285201234", "2325205678"],
+ "orders" : [{"o_id": 21, "o_shop": "Meno Park 1st", "o_amount": 192.5,
+ "items" : [ {"i_name" : "beef", "i_number": 22, "i_supplier": "montana"},
+ {"i_name" : "tooth paste", "i_number": 21, "i_supplier": "washington"},
+ {"i_name" : "hat", "i_number": 20, "i_supplier": "california"}
+ ]
+
+ },
+
+ {"o_id": 22, "o_shop": "Mountain View 1st", "o_amount": 680.9,
+ "items" : [ {"i_name" : "paper towel", "i_number": 23, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 24, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 25, "i_supplier": "california"}
+ ]
+ },
+
+ {"o_id": 23, "o_shop": "Sunnyvale 1st", "o_amount": 772.2,
+ "items" : [ {"i_name" : "paper towel", "i_number": 26, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 27, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 28, "i_supplier": "california"}
+ ]
+ }
+
+ ],
+ "c_address" : {
+ "Area": "bay area, CA",
+ "City": "Milpitas",
+ "State": "CA",
+ "c_phone": [{"area_code" : "120", "phone": "5200001"}, {"area_code": "728", "phone": "5201234"}, {"area_code": "232", "phone" :"5205678"}]
+ }
+}
+{
+ "c_name" : "customer4",
+ "c_id" : 4,
+ "c_phone" : ["6509200001", "4088201234", "6127205678"],
+ "orders" : [{"o_id": 30, "o_shop": "Mountain View 1st", "o_amount": 870.2,
+ "items" : [ {"i_name" : "beef", "i_number": 32, "i_supplier": "montana"},
+ {"i_name" : "tooth paste", "i_number": 31, "i_supplier": "washington"},
+ {"i_name" : "hat", "i_number": 30, "i_supplier": "california"}
+ ]
+
+ },
+
+ {"o_id": 31, "o_shop": "Sunnyvale 1st", "o_amount": 970.5,
+ "items" : [ {"i_name" : "beef", "i_number": 32, "i_supplier": "montana"},
+ {"i_name" : "tooth paste", "i_number": 31, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 30, "i_supplier": "california"}
+ ]
+
+ },
+
+ {"o_id": 32, "o_shop": "Meno Park 1st", "o_amount": 1030.1,
+ "items" : [ {"i_name" : "paper towel", "i_number": 36, "i_supplier": "oregan"},
+ {"i_name" : "tooth paste", "i_number": 37, "i_supplier": "washington"},
+ {"i_name" : "cheese", "i_number": 38, "i_supplier": "california"}
+ ]
+ }
+
+ ],
+ "c_address" : {
+ "Area": "Greater Sandiego",
+ "City": "Sandiego",
+ "State": "CA",
+ "c_phone": [{"area_code": "650", "phone": "9200001"}, {"area_code": "408", "phone": "8201234"}, {"area_code": "612", "phone": "7205678"}]
+ }
+}
+