PHOENIX-1890 Provide queries for adding/deleting jars to/from common place in hdfs which is used by dynamic class loader(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f006df54 Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f006df54 Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f006df54 Branch: refs/heads/calcite Commit: f006df5451859eb9d22130bb46b58460eee49674 Parents: 236ce1c Author: Rajeshbabu Chintaguntla <[email protected]> Authored: Thu Jul 16 23:18:38 2015 +0530 Committer: Rajeshbabu Chintaguntla <[email protected]> Committed: Thu Jul 16 23:18:38 2015 +0530 ---------------------------------------------------------------------- .../phoenix/end2end/UserDefinedFunctionsIT.java | 67 ++++-- phoenix-core/src/main/antlr3/PhoenixSQL.g | 23 ++ .../phoenix/compile/ListJarsQueryPlan.java | 216 +++++++++++++++++++ .../apache/phoenix/jdbc/PhoenixStatement.java | 175 +++++++++++++++ .../apache/phoenix/parse/AddJarsStatement.java | 38 ++++ .../phoenix/parse/DeleteJarStatement.java | 19 ++ .../apache/phoenix/parse/ListJarsStatement.java | 34 +++ .../apache/phoenix/parse/ParseNodeFactory.java | 12 ++ 8 files changed, 564 insertions(+), 20 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/it/java/org/apache/phoenix/end2end/UserDefinedFunctionsIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UserDefinedFunctionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UserDefinedFunctionsIT.java index e2b7b4c..cd1e380 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UserDefinedFunctionsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UserDefinedFunctionsIT.java @@ -207,9 +207,6 @@ public class UserDefinedFunctionsIT extends BaseOwnClusterIT{ conf.set(DYNAMIC_JARS_DIR_KEY, string+"/hbase/tmpjars"); util.startMiniHBaseCluster(1, 1); UDFExpression.setConfig(conf); - compileTestClass(MY_REVERSE_CLASS_NAME, MY_REVERSE_PROGRAM, 1); - compileTestClass(MY_SUM_CLASS_NAME, MY_SUM_PROGRAM, 2); - compileTestClass(MY_ARRAY_INDEX_CLASS_NAME, MY_ARRAY_INDEX_PROGRAM, 3); String clientPort = util.getConfiguration().get(QueryServices.ZOOKEEPER_PORT_ATTRIB); url = @@ -217,10 +214,54 @@ public class UserDefinedFunctionsIT extends BaseOwnClusterIT{ + clientPort + JDBC_PROTOCOL_TERMINATOR + PHOENIX_TEST_DRIVER_URL_PARAM; Map<String, String> props = Maps.newHashMapWithExpectedSize(1); props.put(QueryServices.ALLOW_USER_DEFINED_FUNCTIONS_ATTRIB, "true"); + props.put(QueryServices.DYNAMIC_JARS_DIR_KEY,string+"/hbase/tmpjars/"); driver = initAndRegisterDriver(url, new ReadOnlyProps(props.entrySet().iterator())); + compileTestClass(MY_REVERSE_CLASS_NAME, MY_REVERSE_PROGRAM, 1); + compileTestClass(MY_SUM_CLASS_NAME, MY_SUM_PROGRAM, 2); + compileTestClass(MY_ARRAY_INDEX_CLASS_NAME, MY_ARRAY_INDEX_PROGRAM, 3); + compileTestClass(MY_ARRAY_INDEX_CLASS_NAME, MY_ARRAY_INDEX_PROGRAM, 4); } @Test + public void testListJars() throws Exception { + Connection conn = driver.connect(url, EMPTY_PROPS); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("list jars"); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar1.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar2.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar3.jar", rs.getString("jar_location")); + assertFalse(rs.next()); + } + + @Test + public void testDeleteJar() throws Exception { + Connection conn = driver.connect(url, EMPTY_PROPS); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("list jars"); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar1.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar2.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar3.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar4.jar", rs.getString("jar_location")); + assertFalse(rs.next()); + stmt.execute("delete jar '"+ util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar4.jar'"); + rs = stmt.executeQuery("list jars"); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar1.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar2.jar", rs.getString("jar_location")); + assertTrue(rs.next()); + assertEquals(util.getConfiguration().get(QueryServices.DYNAMIC_JARS_DIR_KEY)+"/"+"myjar3.jar", rs.getString("jar_location")); + assertFalse(rs.next()); + } + + @Test public void testCreateFunction() throws Exception { Connection conn = driver.connect(url, EMPTY_PROPS); Statement stmt = conn.createStatement(); @@ -776,23 +817,9 @@ public class UserDefinedFunctionsIT extends BaseOwnClusterIT{ jarFos.close(); assertTrue(jarFile.exists()); - - InputStream inputStream = new BufferedInputStream(new FileInputStream(jarPath)); - FileSystem fs = util.getDefaultRootDirPath().getFileSystem(util.getConfiguration()); - Path jarsLocation = new Path(util.getConfiguration().get(DYNAMIC_JARS_DIR_KEY)); - Path myJarPath; - if (jarsLocation.toString().endsWith("/")) { - myJarPath = new Path(jarsLocation.toString() + jarName); - } else { - myJarPath = new Path(jarsLocation.toString() + "/" + jarName); - } - OutputStream outputStream = fs.create(myJarPath); - try { - IOUtils.copyBytes(inputStream, outputStream, 4096, false); - } finally { - IOUtils.closeStream(inputStream); - IOUtils.closeStream(outputStream); - } + Connection conn = driver.connect(url, EMPTY_PROPS); + Statement stmt = conn.createStatement(); + stmt.execute("add jars '"+jarFile.getAbsolutePath()+"'"); } finally { if (javaFile != null) javaFile.delete(); if (classFile != null) classFile.delete(); http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/antlr3/PhoenixSQL.g ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g index 10fda68..9fde48f 100644 --- a/phoenix-core/src/main/antlr3/PhoenixSQL.g +++ b/phoenix-core/src/main/antlr3/PhoenixSQL.g @@ -123,6 +123,8 @@ tokens DEFAULTVALUE='defaultvalue'; CONSTANT = 'constant'; REPLACE = 'replace'; + LIST = 'list'; + JARS='jars'; } @@ -182,6 +184,7 @@ import org.apache.phoenix.schema.types.PUnsignedTimestamp; import org.apache.phoenix.util.SchemaUtil; import org.apache.phoenix.parse.LikeParseNode.LikeType; import org.apache.phoenix.trace.util.Tracing; +import org.apache.phoenix.parse.AddJarsStatement; } @lexer::header { @@ -398,6 +401,9 @@ oneStatement returns [BindableStatement ret] | s=trace_node | s=create_function_node | s=drop_function_node + | s=add_jars_node + | s=list_jars_node + | s=delete_jar_node | s=alter_session_node | s=create_sequence_node | s=drop_sequence_node @@ -558,6 +564,18 @@ drop_function_node returns [DropFunctionStatement ret] : DROP FUNCTION (IF ex=EXISTS)? function=identifier {$ret = factory.dropFunction(SchemaUtil.normalizeIdentifier(function), ex!=null);} ; +add_jars_node returns [AddJarsStatement ret] + : ADD JARS jarPaths = one_or_more_jarpaths { $ret = factory.addJars(jarPaths);} + ; + +list_jars_node returns [ListJarsStatement ret] + : LIST JARS { $ret = factory.listJars();} + ; + +delete_jar_node returns [DeleteJarStatement ret] + : DELETE JAR jarPath = jar_path { $ret = factory.deleteJar(jarPath);} + ; + // Parse an alter session statement. alter_session_node returns [AlterSessionStatement ret] : ALTER SESSION (SET p=properties) @@ -914,6 +932,11 @@ one_or_more_expressions returns [List<ParseNode> ret] : e = expression {$ret.add(e);} (COMMA e = expression {$ret.add(e);} )* ; +one_or_more_jarpaths returns [List<LiteralParseNode> ret] +@init{ret = new ArrayList<LiteralParseNode>(); } + : jarPath = jar_path {$ret.add(jarPath);} (COMMA jarPath = jar_path {$ret.add(jarPath);} )* + ; + zero_or_more_expressions returns [List<ParseNode> ret] @init{ret = new ArrayList<ParseNode>(); } : (v = expression {$ret.add(v);})? (COMMA v = expression {$ret.add(v);} )* http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java new file mode 100644 index 0000000..9fdf35b --- /dev/null +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java @@ -0,0 +1,216 @@ +package org.apache.phoenix.compile; + +import java.io.IOException; +import java.sql.ParameterMetaData; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocatedFileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RemoteIterator; +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.KeyValue.Type; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.phoenix.compile.GroupByCompiler.GroupBy; +import org.apache.phoenix.compile.OrderByCompiler.OrderBy; +import org.apache.phoenix.expression.Determinism; +import org.apache.phoenix.expression.Expression; +import org.apache.phoenix.expression.LiteralExpression; +import org.apache.phoenix.expression.RowKeyColumnExpression; +import org.apache.phoenix.iterate.DefaultParallelScanGrouper; +import org.apache.phoenix.iterate.ParallelScanGrouper; +import org.apache.phoenix.iterate.ResultIterator; +import org.apache.phoenix.jdbc.PhoenixParameterMetaData; +import org.apache.phoenix.jdbc.PhoenixStatement; +import org.apache.phoenix.parse.FilterableStatement; +import org.apache.phoenix.parse.LiteralParseNode; +import org.apache.phoenix.parse.ParseNodeFactory; +import org.apache.phoenix.query.HBaseFactoryProvider; +import org.apache.phoenix.query.KeyRange; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.schema.PColumn; +import org.apache.phoenix.schema.PColumnImpl; +import org.apache.phoenix.schema.PNameFactory; +import org.apache.phoenix.schema.RowKeyValueAccessor; +import org.apache.phoenix.schema.SortOrder; +import org.apache.phoenix.schema.TableRef; +import org.apache.phoenix.schema.tuple.ResultTuple; +import org.apache.phoenix.schema.tuple.Tuple; +import org.apache.phoenix.schema.types.PVarchar; +import org.apache.phoenix.util.ByteUtil; +import org.apache.phoenix.util.SizedUtil; + +public class ListJarsQueryPlan implements QueryPlan { + + private PhoenixStatement stmt = null; + private StatementContext context = null; + private boolean first = true; + + private static final RowProjector JARS_PROJECTOR; + + static { + List<ExpressionProjector> projectedColumns = new ArrayList<ExpressionProjector>(); + PColumn column = + new PColumnImpl(PNameFactory.newName("jar_location"), null, + PVarchar.INSTANCE, null, null, false, 0, SortOrder.getDefault(), 0, null, + false, null); + List<PColumn> columns = new ArrayList<PColumn>(); + columns.add(column); + Expression expression = + new RowKeyColumnExpression(column, new RowKeyValueAccessor(columns, 0)); + projectedColumns.add(new ExpressionProjector("jar_location", "", expression, + true)); + int estimatedByteSize = SizedUtil.KEY_VALUE_SIZE; + JARS_PROJECTOR = new RowProjector(projectedColumns, estimatedByteSize, false); + } + + public ListJarsQueryPlan(PhoenixStatement stmt) { + this.stmt = stmt; + this.context = new StatementContext(stmt); + } + + @Override + public StatementContext getContext() { + return this.context; + } + + @Override + public ParameterMetaData getParameterMetaData() { + return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA; + } + + @Override + public ExplainPlan getExplainPlan() throws SQLException { + return ExplainPlan.EMPTY_PLAN; + } + + @Override + public ResultIterator iterator() throws SQLException { + return iterator(DefaultParallelScanGrouper.getInstance()); + } + + @Override + public ResultIterator iterator(ParallelScanGrouper scanGrouper) throws SQLException { + return new ResultIterator() { + private RemoteIterator<LocatedFileStatus> listFiles = null; + + @Override + public void close() throws SQLException { + + } + + @Override + public Tuple next() throws SQLException { + try { + if(first) { + String dynamicJarsDir = + stmt.getConnection().getQueryServices().getProps() + .get(QueryServices.DYNAMIC_JARS_DIR_KEY); + if(dynamicJarsDir == null) { + throw new SQLException(QueryServices.DYNAMIC_JARS_DIR_KEY + + " is not configured for the listing the jars."); + } + dynamicJarsDir = + dynamicJarsDir.endsWith("/") ? dynamicJarsDir : dynamicJarsDir + '/'; + Configuration conf = HBaseFactoryProvider.getConfigurationFactory().getConfiguration(); + Path dynamicJarsDirPath = new Path(dynamicJarsDir); + FileSystem fs = dynamicJarsDirPath.getFileSystem(conf); + listFiles = fs.listFiles(dynamicJarsDirPath, true); + first = false; + } + if(listFiles == null || !listFiles.hasNext()) return null; + ImmutableBytesWritable ptr = new ImmutableBytesWritable(); + ParseNodeFactory factory = new ParseNodeFactory(); + LiteralParseNode literal = + factory.literal(listFiles.next().getPath().toString()); + LiteralExpression expression = + LiteralExpression.newConstant(literal.getValue(), PVarchar.INSTANCE, + Determinism.ALWAYS); + expression.evaluate(null, ptr); + byte[] rowKey = ByteUtil.copyKeyBytesIfNecessary(ptr); + Cell cell = + CellUtil.createCell(rowKey, HConstants.EMPTY_BYTE_ARRAY, + HConstants.EMPTY_BYTE_ARRAY, System.currentTimeMillis(), + Type.Put.getCode(), HConstants.EMPTY_BYTE_ARRAY); + List<Cell> cells = new ArrayList<Cell>(1); + cells.add(cell); + return new ResultTuple(Result.create(cells)); + } catch (IOException e) { + throw new SQLException(e); + } + } + + @Override + public void explain(List<String> planSteps) { + } + }; + } + + @Override + public long getEstimatedSize() { + return PVarchar.INSTANCE.getByteSize(); + } + + @Override + public TableRef getTableRef() { + return null; + } + + @Override + public RowProjector getProjector() { + return JARS_PROJECTOR; + } + + @Override + public Integer getLimit() { + return null; + } + + @Override + public OrderBy getOrderBy() { + return OrderBy.EMPTY_ORDER_BY; + } + + @Override + public GroupBy getGroupBy() { + return GroupBy.EMPTY_GROUP_BY; + } + + @Override + public List<KeyRange> getSplits() { + return Collections.emptyList(); + } + + @Override + public List<List<Scan>> getScans() { + return Collections.emptyList(); + } + + @Override + public FilterableStatement getStatement() { + return null; + } + + @Override + public boolean isDegenerate() { + return false; + } + + @Override + public boolean isRowKeyOrdered() { + return false; + } + + @Override + public boolean useRoundRobinIterator() { + return false; + } +} http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java index 9689589..056263a 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java @@ -21,6 +21,7 @@ import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_ import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIME; import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SELECT_SQL_COUNTER; +import java.io.File; import java.io.IOException; import java.io.Reader; import java.sql.ParameterMetaData; @@ -36,6 +37,11 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocatedFileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RemoteIterator; import org.apache.hadoop.hbase.client.Consistency; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.util.Pair; @@ -52,6 +58,7 @@ import org.apache.phoenix.compile.ExplainPlan; import org.apache.phoenix.compile.ExpressionProjector; import org.apache.phoenix.compile.FromCompiler; import org.apache.phoenix.compile.GroupByCompiler.GroupBy; +import org.apache.phoenix.compile.ListJarsQueryPlan; import org.apache.phoenix.compile.MutationPlan; import org.apache.phoenix.compile.OrderByCompiler.OrderBy; import org.apache.phoenix.compile.QueryCompiler; @@ -74,6 +81,7 @@ import org.apache.phoenix.iterate.MaterializedResultIterator; import org.apache.phoenix.iterate.ParallelScanGrouper; import org.apache.phoenix.iterate.ResultIterator; import org.apache.phoenix.parse.AddColumnStatement; +import org.apache.phoenix.parse.AddJarsStatement; import org.apache.phoenix.parse.AliasedNode; import org.apache.phoenix.parse.AlterIndexStatement; import org.apache.phoenix.parse.AlterSessionStatement; @@ -84,6 +92,7 @@ import org.apache.phoenix.parse.CreateFunctionStatement; import org.apache.phoenix.parse.CreateIndexStatement; import org.apache.phoenix.parse.CreateSequenceStatement; import org.apache.phoenix.parse.CreateTableStatement; +import org.apache.phoenix.parse.DeleteJarStatement; import org.apache.phoenix.parse.DeleteStatement; import org.apache.phoenix.parse.DropColumnStatement; import org.apache.phoenix.parse.DropFunctionStatement; @@ -95,6 +104,8 @@ import org.apache.phoenix.parse.FilterableStatement; import org.apache.phoenix.parse.HintNode; import org.apache.phoenix.parse.IndexKeyConstraint; import org.apache.phoenix.parse.LimitNode; +import org.apache.phoenix.parse.ListJarsStatement; +import org.apache.phoenix.parse.LiteralParseNode; import org.apache.phoenix.parse.NamedNode; import org.apache.phoenix.parse.NamedTableNode; import org.apache.phoenix.parse.OrderByNode; @@ -110,6 +121,7 @@ import org.apache.phoenix.parse.TraceStatement; import org.apache.phoenix.parse.UDFParseNode; import org.apache.phoenix.parse.UpdateStatisticsStatement; import org.apache.phoenix.parse.UpsertStatement; +import org.apache.phoenix.query.HBaseFactoryProvider; import org.apache.phoenix.query.KeyRange; import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.query.QueryServices; @@ -631,6 +643,152 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho } } + private static class ExecutableAddJarsStatement extends AddJarsStatement implements CompilableStatement { + + public ExecutableAddJarsStatement(List<LiteralParseNode> jarPaths) { + super(jarPaths); + } + + + @SuppressWarnings("unchecked") + @Override + public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException { + final StatementContext context = new StatementContext(stmt); + return new MutationPlan() { + + @Override + public StatementContext getContext() { + return context; + } + + @Override + public ParameterMetaData getParameterMetaData() { + return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA; + } + + @Override + public ExplainPlan getExplainPlan() throws SQLException { + return new ExplainPlan(Collections.singletonList("ADD JARS")); + } + + @Override + public PhoenixConnection getConnection() { + return stmt.getConnection(); + } + + @Override + public MutationState execute() throws SQLException { + String dynamicJarsDir = stmt.getConnection().getQueryServices().getProps().get(QueryServices.DYNAMIC_JARS_DIR_KEY); + if(dynamicJarsDir == null) { + throw new SQLException(QueryServices.DYNAMIC_JARS_DIR_KEY+" is not configured for placing the jars."); + } + dynamicJarsDir = + dynamicJarsDir.endsWith("/") ? dynamicJarsDir : dynamicJarsDir + '/'; + Configuration conf = HBaseFactoryProvider.getConfigurationFactory().getConfiguration(); + Path dynamicJarsDirPath = new Path(dynamicJarsDir); + for (LiteralParseNode jarPath : getJarPaths()) { + String jarPathStr = (String)jarPath.getValue(); + if(!jarPathStr.endsWith(".jar")) { + throw new SQLException(jarPathStr + " is not a valid jar file path."); + } + } + + try { + FileSystem fs = dynamicJarsDirPath.getFileSystem(conf); + List<LiteralParseNode> jarPaths = getJarPaths(); + for (LiteralParseNode jarPath : jarPaths) { + File f = new File((String) jarPath.getValue()); + fs.copyFromLocalFile(new Path(f.getAbsolutePath()), new Path( + dynamicJarsDir + f.getName())); + } + } catch(IOException e) { + throw new SQLException(e); + } + return new MutationState(0, context.getConnection()); + } + }; + + } + } + + private static class ExecutableDeleteJarStatement extends DeleteJarStatement implements CompilableStatement { + + public ExecutableDeleteJarStatement(LiteralParseNode jarPath) { + super(jarPath); + } + + + @SuppressWarnings("unchecked") + @Override + public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException { + final StatementContext context = new StatementContext(stmt); + return new MutationPlan() { + + @Override + public StatementContext getContext() { + return context; + } + + @Override + public ParameterMetaData getParameterMetaData() { + return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA; + } + + @Override + public ExplainPlan getExplainPlan() throws SQLException { + return new ExplainPlan(Collections.singletonList("DELETE JAR")); + } + + @Override + public PhoenixConnection getConnection() { + return stmt.getConnection(); + } + + @Override + public MutationState execute() throws SQLException { + String dynamicJarsDir = stmt.getConnection().getQueryServices().getProps().get(QueryServices.DYNAMIC_JARS_DIR_KEY); + if (dynamicJarsDir == null) { + throw new SQLException(QueryServices.DYNAMIC_JARS_DIR_KEY + + " is not configured."); + } + dynamicJarsDir = + dynamicJarsDir.endsWith("/") ? dynamicJarsDir : dynamicJarsDir + '/'; + Configuration conf = HBaseFactoryProvider.getConfigurationFactory().getConfiguration(); + Path dynamicJarsDirPath = new Path(dynamicJarsDir); + try { + FileSystem fs = dynamicJarsDirPath.getFileSystem(conf); + String jarPathStr = (String)getJarPath().getValue(); + if(!jarPathStr.endsWith(".jar")) { + throw new SQLException(jarPathStr + " is not a valid jar file path."); + } + Path p = new Path(jarPathStr); + if(fs.exists(p)) { + fs.delete(p, false); + } + } catch(IOException e) { + throw new SQLException(e); + } + return new MutationState(0, context.getConnection()); + } + }; + + } + } + + private static class ExecutableListJarsStatement extends ListJarsStatement implements CompilableStatement { + + public ExecutableListJarsStatement() { + super(); + } + + + @SuppressWarnings("unchecked") + @Override + public QueryPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException { + return new ListJarsQueryPlan(stmt); + } + } + private static class ExecutableCreateIndexStatement extends CreateIndexStatement implements CompilableStatement { public ExecutableCreateIndexStatement(NamedNode indexName, NamedTableNode dataTable, IndexKeyConstraint ikConstraint, List<ColumnName> includeColumns, List<ParseNode> splits, @@ -1026,6 +1184,23 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho public CreateFunctionStatement createFunction(PFunction functionInfo, boolean temporary, boolean isReplace) { return new ExecutableCreateFunctionStatement(functionInfo, temporary, isReplace); } + + @Override + public AddJarsStatement addJars(List<LiteralParseNode> jarPaths) { + return new ExecutableAddJarsStatement(jarPaths); + } + + @Override + public DeleteJarStatement deleteJar(LiteralParseNode jarPath) { + return new ExecutableDeleteJarStatement(jarPath); + } + + @Override + public ListJarsStatement listJars() { + return new ExecutableListJarsStatement(); + } + + @Override public DropSequenceStatement dropSequence(TableName tableName, boolean ifExists, int bindCount){ return new ExecutableDropSequenceStatement(tableName, ifExists, bindCount); http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/parse/AddJarsStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/AddJarsStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/AddJarsStatement.java new file mode 100644 index 0000000..b1eeea6 --- /dev/null +++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/AddJarsStatement.java @@ -0,0 +1,38 @@ +/* + * 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.phoenix.parse; + +import java.util.List; + +public class AddJarsStatement extends MutableStatement { + + List<LiteralParseNode> jarPaths; + + public AddJarsStatement(List<LiteralParseNode> jarPaths) { + this.jarPaths = jarPaths; + } + + @Override + public int getBindCount() { + return 0; + } + + public List<LiteralParseNode> getJarPaths() { + return jarPaths; + } +} http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/parse/DeleteJarStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/DeleteJarStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/DeleteJarStatement.java new file mode 100644 index 0000000..7ccdc72 --- /dev/null +++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/DeleteJarStatement.java @@ -0,0 +1,19 @@ +package org.apache.phoenix.parse; + +public class DeleteJarStatement extends MutableStatement { + + private LiteralParseNode jarPath; + + public DeleteJarStatement(LiteralParseNode jarPath) { + this.jarPath = jarPath; + } + + @Override + public int getBindCount() { + return 0; + } + + public LiteralParseNode getJarPath() { + return jarPath; + } +} http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/parse/ListJarsStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ListJarsStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ListJarsStatement.java new file mode 100644 index 0000000..e9821fb --- /dev/null +++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ListJarsStatement.java @@ -0,0 +1,34 @@ +/* + * 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.phoenix.parse; + +import org.apache.phoenix.jdbc.PhoenixStatement.Operation; + +public class ListJarsStatement implements BindableStatement { + + @Override + public int getBindCount() { + return 0; + } + + @Override + public Operation getOperation() { + return Operation.QUERY; + } + +} http://git-wip-us.apache.org/repos/asf/phoenix/blob/f006df54/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java index 49b14c6..cd239ac 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java @@ -301,6 +301,18 @@ public class ParseNodeFactory { return new CreateFunctionStatement(functionInfo, temporary, isReplace); } + public AddJarsStatement addJars(List<LiteralParseNode> jarPaths) { + return new AddJarsStatement(jarPaths); + } + + public ListJarsStatement listJars() { + return new ListJarsStatement(); + } + + public DeleteJarStatement deleteJar(LiteralParseNode jarPath) { + return new DeleteJarStatement(jarPath); + } + public DropFunctionStatement dropFunction(String functionName, boolean ifExists) { return new DropFunctionStatement(functionName, ifExists); }
