hawq-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sansanichfb <...@git.apache.org>
Subject [GitHub] incubator-hawq pull request #1353: HAWQ-1605. Support INSERT in PXF JDBC plu...
Date Thu, 19 Apr 2018 01:43:04 GMT
Github user sansanichfb commented on a diff in the pull request:

    https://github.com/apache/incubator-hawq/pull/1353#discussion_r182614960
  
    --- Diff: pxf/pxf-jdbc/src/main/java/org/apache/hawq/pxf/plugins/jdbc/JdbcAccessor.java
---
    @@ -0,0 +1,469 @@
    +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.OneRow;
    +import org.apache.hawq.pxf.api.OneField;
    +import org.apache.hawq.pxf.api.ReadAccessor;
    +import org.apache.hawq.pxf.api.WriteAccessor;
    +import org.apache.hawq.pxf.api.io.DataType;
    +import org.apache.hawq.pxf.api.UserDataException;
    +import org.apache.hawq.pxf.api.utilities.ColumnDescriptor;
    +import org.apache.hawq.pxf.api.utilities.InputData;
    +
    +import java.util.List;
    +import java.io.IOException;
    +import java.text.ParseException;
    +import java.math.BigDecimal;
    +import java.sql.Types;
    +import java.sql.Date;
    +import java.sql.PreparedStatement;
    +import java.sql.ResultSet;
    +import java.sql.SQLException;
    +import java.sql.SQLTimeoutException;
    +import java.sql.Statement;
    +import java.sql.Timestamp;
    +
    +import org.apache.commons.logging.Log;
    +import org.apache.commons.logging.LogFactory;
    +
    +/**
    + * JDBC tables accessor
    + *
    + * The SELECT queries are processed by {@link java.sql.Statement}
    + *
    + * The INSERT queries are processed by {@link java.sql.PreparedStatement} and
    + * built-in JDBC batches of arbitrary size
    + */
    +public class JdbcAccessor extends JdbcPlugin implements ReadAccessor, WriteAccessor {
    +    /**
    +     * Class constructor
    +     */
    +    public JdbcAccessor(InputData inputData) throws UserDataException {
    +        super(inputData);
    +    }
    +
    +    /**
    +     * openForRead() implementation
    +     * Create query, open JDBC connection, execute query and store the result into resultSet
    +     *
    +     * @throws SQLException if a database access error occurs
    +     * @throws SQLTimeoutException if a problem with the connection occurs
    +     * @throws ParseException if th SQL statement provided in PXF InputData is incorrect
    +     * @throws ClassNotFoundException if the superclass implementation disappeared
    +     */
    +    @Override
    +    public boolean openForRead() throws SQLException, SQLTimeoutException, ParseException,
ClassNotFoundException {
    +        if (statementRead != null && !statementRead.isClosed()) {
    +            return true;
    +        }
    +
    +        super.openConnection();
    +
    +        queryRead = buildSelectQuery();
    +        statementRead = dbConn.createStatement();
    +        resultSetRead = statementRead.executeQuery(queryRead);
    +
    +        return true;
    +    }
    +
    +    /**
    +     * readNextObject() implementation
    +     * Retreive the next tuple from resultSet and return it
    +     *
    +     * @throws SQLException if a problem in resultSet occurs
    +     */
    +    @Override
    +    public OneRow readNextObject() throws SQLException {
    +        if (resultSetRead.next()) {
    +            return new OneRow(resultSetRead);
    +        }
    +        return null;
    +    }
    +
    +    /**
    +     * closeForRead() implementation
    +     *
    +     * @throws SQLException if a database access error occurs
    +     */
    +    @Override
    +    public void closeForRead() throws SQLException {
    +        if (statementRead != null && !statementRead.isClosed()) {
    +            statementRead.close();
    +            statementRead = null;
    +        }
    +        super.closeConnection();
    +    }
    +
    +    /**
    +     * openForWrite() implementation
    +     * Create query template and open JDBC connection
    +     *
    +     * @throws SQLException if a database access error occurs
    +     * @throws SQLTimeoutException if a problem with the connection occurs
    +     * @throws ParseException if the SQL statement provided in PXF InputData is incorrect
    +     * @throws ClassNotFoundException if the superclass implementation has disappeared
    +     */
    +    @Override
    +    public boolean openForWrite() throws SQLException, SQLTimeoutException, ParseException,
ClassNotFoundException {
    +        if (statementWrite != null && !statementWrite.isClosed()) {
    +            return true;
    +        }
    +
    +        super.openConnection();
    +        if (dbMeta.supportsTransactions()) {
    +            dbConn.setAutoCommit(false);
    +        }
    +
    +        queryWrite = buildInsertQuery();
    +        statementWrite = dbConn.prepareStatement(queryWrite);
    +
    +        if ((batchSize != 0) && (!dbMeta.supportsBatchUpdates())) {
    +            LOG.info(
    +                "The database '" +
    +                dbMeta.getDatabaseProductName() +
    +                "' does not support batch updates. The current request will be handled
without batching"
    +            );
    +            batchSize = 0;
    +        }
    +
    +        return true;
    +    }
    +
    +	/**
    +     * writeNextObject() implementation
    +     *
    +     * If batchSize is not 0 or 1, add a tuple to the batch of statementWrite
    +     * Otherwise, execute an INSERT query immediately
    +     *
    +     * In both cases, a {@link java.sql.PreparedStatement} is used
    +     *
    +     * @throws SQLException if a database access error occurs
    +     * @throws IOException if the data provided by {@link JdbcResolver} is corrupted
    +     */
    +    @Override
    +    @SuppressWarnings("unchecked")
    +    public boolean writeNextObject(OneRow row) throws SQLException, IOException {
    +        // This cast is safe because the data in the row is formed by JdbcPlugin
    +        List<OneField> tuple = (List<OneField>) row.getData();
    +
    +        if (LOG.isDebugEnabled()) {
    +            LOG.debug("writeNextObject() called");
    +        }
    +
    +        for (int i = 1; i <= tuple.size(); i++) {
    +            OneField field = tuple.get(i - 1);
    +            if (LOG.isDebugEnabled()) {
    +                LOG.debug("Field " + i + ": " + DataType.get(field.type).toString());
    +            }
    +            switch (DataType.get(field.type)) {
    +                case INTEGER:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.INTEGER);
    +                    }
    +                    else {
    +                        statementWrite.setInt(i, (int)field.val);
    +                    }
    +                    break;
    +                case BIGINT:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.INTEGER);
    +                    }
    +                    else {
    +                        statementWrite.setLong(i, (long)field.val);
    +                    }
    +                    break;
    +                case SMALLINT:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.INTEGER);
    +                    }
    +                    else {
    +                        statementWrite.setShort(i, (short)field.val);
    +                    }
    +                    break;
    +                case REAL:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.FLOAT);
    +                    }
    +                    else {
    +                        statementWrite.setFloat(i, (float)field.val);
    +                    }
    +                    break;
    +                case FLOAT8:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.DOUBLE);
    +                    }
    +                    else {
    +                        statementWrite.setDouble(i, (double)field.val);
    +                    }
    +                    break;
    +                case BOOLEAN:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.BOOLEAN);
    +                    }
    +                    else {
    +                        statementWrite.setBoolean(i, (boolean)field.val);
    +                    }
    +                    break;
    +                case NUMERIC:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.NUMERIC);
    +                    }
    +                    else {
    +                        statementWrite.setBigDecimal(i, (BigDecimal)field.val);
    +                    }
    +                    break;
    +                case VARCHAR:
    +                case BPCHAR:
    +                case TEXT:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.VARCHAR);
    +                    }
    +                    else {
    +                        statementWrite.setString(i, (String)field.val);
    +                    }
    +                    break;
    +                case BYTEA:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.BINARY);
    +                    }
    +                    else {
    +                        statementWrite.setBytes(i, (byte[])field.val);
    +                    }
    +                    break;
    +                case TIMESTAMP:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.TIMESTAMP);
    +                    }
    +                    else {
    +                        statementWrite.setTimestamp(i, (Timestamp)field.val);
    +                    }
    +                    break;
    +                case DATE:
    +                    if (field.val == null) {
    +                        statementWrite.setNull(i, Types.DATE);
    +                    }
    +                    else {
    +                        statementWrite.setDate(i, (Date)field.val);
    +                    }
    +                    break;
    +                default:
    +                    throw new IOException("The data tuple from JdbcResolver is corrupted");
    --- End diff --
    
    This message is not very actionable. I'd be more specific and mention that certain data
type is not supported. 


---

Mime
View raw message