pxf/pxf-jdbc/src/main/java/org/apache/hawq/pxf/plugins/jdbc/JdbcPlugin.java - hawq - Git at Google

 package org.apache.hawq.pxf.plugins.jdbc;

 /*
  * 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.
  */

 import org.apache.hawq.pxf.api.UserDataException;
 import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
 import org.apache.hawq.pxf.api.utilities.InputData;
 import org.apache.hawq.pxf.api.utilities.Plugin;

 import java.util.ArrayList;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.SQLTimeoutException;
 import java.sql.Statement;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 /**
  * JDBC tables plugin (base class)
  *
  * Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}.
  */
 public class JdbcPlugin extends Plugin {
     /**
      * Class constructor
      *
      * @param input {@link InputData} provided by PXF
      *
      * @throws UserDataException if one of the required request parameters is not set
      */
     public JdbcPlugin(InputData input) throws UserDataException {
         super(input);

         jdbcDriver = input.getUserProperty("JDBC_DRIVER");
         if (jdbcDriver == null) {
             throw new UserDataException("JDBC_DRIVER is a required parameter");
         }

         dbUrl = input.getUserProperty("DB_URL");
         if (dbUrl == null) {
             throw new UserDataException("DB_URL is a required parameter");
         }

         tableName = input.getDataSource();
         if (tableName == null) {
             throw new UserDataException("Data source must be provided");
         }
         /*
         At the moment, when writing into some table, the table name is
         concatenated with a special string that is necessary to write into HDFS.
         However, a raw table name is necessary in case of JDBC.
         The correct table name is extracted here.
         */
         Matcher matcher = tableNamePattern.matcher(tableName);
         if (matcher.matches()) {
             inputData.setDataSource(matcher.group(1));
             tableName = input.getDataSource();
         }

         columns = inputData.getTupleDescription();
         if (columns == null) {
             throw new UserDataException("Tuple description must be provided");
         }

         // This parameter is not required. The default value is null
         user = input.getUserProperty("USER");
         if (user != null) {
             pass = input.getUserProperty("PASS");
         }

         // This parameter is not required. The default value is 0
         String batchSizeRaw = input.getUserProperty("BATCH_SIZE");
         if (batchSizeRaw != null) {
             try {
                 batchSize = Integer.parseInt(batchSizeRaw);
                 if (batchSize < 1) {
                     throw new NumberFormatException();
                 } else if (batchSize == 0) {
                     batchSize = 1;
                 }
                 batchSizeIsSetByUser = true;
             }
             catch (NumberFormatException e) {
                 throw new UserDataException("BATCH_SIZE is incorrect: must be a non-negative integer");
             }
         }

         // This parameter is not required. The default value is 1
         String poolSizeRaw = input.getUserProperty("POOL_SIZE");
         if (poolSizeRaw != null) {
             try {
                 poolSize = Integer.parseInt(poolSizeRaw);
             }
             catch (NumberFormatException e) {
                 throw new UserDataException("POOL_SIZE is incorrect: must be an integer");
             }
         }
     }

     /**
      * Open a new JDBC connection
      *
      * @throws ClassNotFoundException if the JDBC driver was not found
      * @throws SQLException if a database access error occurs
      * @throws SQLTimeoutException if a connection problem occurs
      */
     public Connection getConnection() throws ClassNotFoundException, SQLException, SQLTimeoutException {
         Connection connection;
         if (LOG.isDebugEnabled()) {
             if (user != null) {
                 LOG.debug(String.format("Open JDBC connection: driver=%s, url=%s, user=%s, pass=%s, table=%s",
                     jdbcDriver, dbUrl, user, pass, tableName));
             }
             else {
                 LOG.debug(String.format("Open JDBC connection: driver=%s, url=%s, table=%s",
                     jdbcDriver, dbUrl, tableName));
             }
         }
         Class.forName(jdbcDriver);
         if (user != null) {
             connection = DriverManager.getConnection(dbUrl, user, pass);
         }
         else {
             connection = DriverManager.getConnection(dbUrl);
         }
         return connection;
     }

     /**
      * Close a JDBC connection
      */
     public static void closeConnection(Connection connection) {
         try {
             if ((connection != null) && (!connection.isClosed())) {
                 if ((connection.getMetaData().supportsTransactions()) && (!connection.getAutoCommit())) {
                     connection.commit();
                 }
                 connection.close();
             }
         }
         catch (SQLException e) {
             LOG.error("JDBC connection close error", e);
         }
     }

     /**
      * Prepare a JDBC PreparedStatement
      *
      * @throws ClassNotFoundException if the JDBC driver was not found
      * @throws SQLException if a database access error occurs
      * @throws SQLTimeoutException if a connection problem occurs
      */
     public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException, SQLTimeoutException, ClassNotFoundException {
         if ((connection == null) || (query == null)) {
             throw new IllegalArgumentException("The provided query or connection is null");
         }
         if (connection.getMetaData().supportsTransactions()) {
             connection.setAutoCommit(false);
         }
         return connection.prepareStatement(query);
     }

     /**
      * Close a JDBC Statement (and the connection it is based on)
      */
     public static void closeStatement(Statement statement) {
         if (statement == null) {
             return;
         }
         Connection connection = null;
         try {
             if (!statement.isClosed()) {
                 connection = statement.getConnection();
                 statement.close();
             }
         }
         catch (Exception e) {}
         closeConnection(connection);
     }

     // JDBC parameters
     protected String jdbcDriver = null;
     protected String dbUrl = null;
     protected String user = null;
     protected String pass = null;

     protected String tableName = null;

     // '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754
     public static final int DEFAULT_BATCH_SIZE = 100;
     // After argument parsing, this value is guaranteed to be >= 1
     protected int batchSize = DEFAULT_BATCH_SIZE;
     protected boolean batchSizeIsSetByUser = false;

     protected int poolSize = 1;

     // Columns description
     protected ArrayList<ColumnDescriptor> columns = null;


     private static final Log LOG = LogFactory.getLog(JdbcPlugin.class);

     // At the moment, when writing into some table, the table name is concatenated with a special string that is necessary to write into HDFS. However, a raw table name is necessary in case of JDBC. This Pattern allows to extract the correct table name from the given InputData.dataSource
     private static final Pattern tableNamePattern = Pattern.compile("/(.*)/[0-9]*-[0-9]*_[0-9]*");
 }
	package org.apache.hawq.pxf.plugins.jdbc;

	/*
	* 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.
	*/

	import org.apache.hawq.pxf.api.UserDataException;
	import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
	import org.apache.hawq.pxf.api.utilities.InputData;
	import org.apache.hawq.pxf.api.utilities.Plugin;

	import java.util.ArrayList;
	import java.util.regex.Matcher;
	import java.util.regex.Pattern;
	import java.sql.Connection;
	import java.sql.DriverManager;
	import java.sql.PreparedStatement;
	import java.sql.SQLException;
	import java.sql.SQLTimeoutException;
	import java.sql.Statement;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	/**
	* JDBC tables plugin (base class)
	*
	* Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}.
	*/
	public class JdbcPlugin extends Plugin {
	/**
	* Class constructor
	*
	* @param input {@link InputData} provided by PXF
	*
	* @throws UserDataException if one of the required request parameters is not set
	*/
	public JdbcPlugin(InputData input) throws UserDataException {
	super(input);

	jdbcDriver = input.getUserProperty("JDBC_DRIVER");
	if (jdbcDriver == null) {
	throw new UserDataException("JDBC_DRIVER is a required parameter");
	}

	dbUrl = input.getUserProperty("DB_URL");
	if (dbUrl == null) {
	throw new UserDataException("DB_URL is a required parameter");
	}

	tableName = input.getDataSource();
	if (tableName == null) {
	throw new UserDataException("Data source must be provided");
	}
	/*
	At the moment, when writing into some table, the table name is
	concatenated with a special string that is necessary to write into HDFS.
	However, a raw table name is necessary in case of JDBC.
	The correct table name is extracted here.
	*/
	Matcher matcher = tableNamePattern.matcher(tableName);
	if (matcher.matches()) {
	inputData.setDataSource(matcher.group(1));
	tableName = input.getDataSource();
	}

	columns = inputData.getTupleDescription();
	if (columns == null) {
	throw new UserDataException("Tuple description must be provided");
	}

	// This parameter is not required. The default value is null
	user = input.getUserProperty("USER");
	if (user != null) {
	pass = input.getUserProperty("PASS");
	}

	// This parameter is not required. The default value is 0
	String batchSizeRaw = input.getUserProperty("BATCH_SIZE");
	if (batchSizeRaw != null) {
	try {
	batchSize = Integer.parseInt(batchSizeRaw);
	if (batchSize < 1) {
	throw new NumberFormatException();
	} else if (batchSize == 0) {
	batchSize = 1;
	}
	batchSizeIsSetByUser = true;
	}
	catch (NumberFormatException e) {
	throw new UserDataException("BATCH_SIZE is incorrect: must be a non-negative integer");
	}
	}

	// This parameter is not required. The default value is 1
	String poolSizeRaw = input.getUserProperty("POOL_SIZE");
	if (poolSizeRaw != null) {
	try {
	poolSize = Integer.parseInt(poolSizeRaw);
	}
	catch (NumberFormatException e) {
	throw new UserDataException("POOL_SIZE is incorrect: must be an integer");
	}
	}
	}

	/**
	* Open a new JDBC connection
	*
	* @throws ClassNotFoundException if the JDBC driver was not found
	* @throws SQLException if a database access error occurs
	* @throws SQLTimeoutException if a connection problem occurs
	*/
	public Connection getConnection() throws ClassNotFoundException, SQLException, SQLTimeoutException {
	Connection connection;
	if (LOG.isDebugEnabled()) {
	if (user != null) {
	LOG.debug(String.format("Open JDBC connection: driver=%s, url=%s, user=%s, pass=%s, table=%s",
	jdbcDriver, dbUrl, user, pass, tableName));
	}
	else {
	LOG.debug(String.format("Open JDBC connection: driver=%s, url=%s, table=%s",
	jdbcDriver, dbUrl, tableName));
	}
	}
	Class.forName(jdbcDriver);
	if (user != null) {
	connection = DriverManager.getConnection(dbUrl, user, pass);
	}
	else {
	connection = DriverManager.getConnection(dbUrl);
	}
	return connection;
	}

	/**
	* Close a JDBC connection
	*/
	public static void closeConnection(Connection connection) {
	try {
	if ((connection != null) && (!connection.isClosed())) {
	if ((connection.getMetaData().supportsTransactions()) && (!connection.getAutoCommit())) {
	connection.commit();
	}
	connection.close();
	}
	}
	catch (SQLException e) {
	LOG.error("JDBC connection close error", e);
	}
	}

	/**
	* Prepare a JDBC PreparedStatement
	*
	* @throws ClassNotFoundException if the JDBC driver was not found
	* @throws SQLException if a database access error occurs
	* @throws SQLTimeoutException if a connection problem occurs
	*/
	public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException, SQLTimeoutException, ClassNotFoundException {
	if ((connection == null) \|\| (query == null)) {
	throw new IllegalArgumentException("The provided query or connection is null");
	}
	if (connection.getMetaData().supportsTransactions()) {
	connection.setAutoCommit(false);
	}
	return connection.prepareStatement(query);
	}

	/**
	* Close a JDBC Statement (and the connection it is based on)
	*/
	public static void closeStatement(Statement statement) {
	if (statement == null) {
	return;
	}
	Connection connection = null;
	try {
	if (!statement.isClosed()) {
	connection = statement.getConnection();
	statement.close();
	}
	}
	catch (Exception e) {}
	closeConnection(connection);
	}

	// JDBC parameters
	protected String jdbcDriver = null;
	protected String dbUrl = null;
	protected String user = null;
	protected String pass = null;

	protected String tableName = null;

	// '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754
	public static final int DEFAULT_BATCH_SIZE = 100;
	// After argument parsing, this value is guaranteed to be >= 1
	protected int batchSize = DEFAULT_BATCH_SIZE;
	protected boolean batchSizeIsSetByUser = false;

	protected int poolSize = 1;

	// Columns description
	protected ArrayList<ColumnDescriptor> columns = null;


	private static final Log LOG = LogFactory.getLog(JdbcPlugin.class);

	// At the moment, when writing into some table, the table name is concatenated with a special string that is necessary to write into HDFS. However, a raw table name is necessary in case of JDBC. This Pattern allows to extract the correct table name from the given InputData.dataSource
	private static final Pattern tableNamePattern = Pattern.compile("/(.)/[0-9]-[0-9]_[0-9]");
	}