phoenix5-hive/src/main/java/org/apache/phoenix/hive/PhoenixMetaHook.java [42:244]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class PhoenixMetaHook implements HiveMetaHook {

    private static final Logger LOG = LoggerFactory.getLogger(PhoenixMetaHook.class);
    private static final String EXTERNAL_TABLE_PURGE = "external.table.purge";

    @Override
    public void preCreateTable(Table table) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Precreate table : " + table.getTableName());
        }

        try (Connection conn = PhoenixConnectionUtil.getConnection(table)) {
            String tableType = table.getTableType();
            String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);

            if (TableType.EXTERNAL_TABLE.name().equals(tableType)) {
                // Check whether phoenix table exists.
                if (!PhoenixUtil.existTable(conn, tableName)) {
                    //For Hive 3.0.0 only external tables are supported. If table doesn't exist,
                    // a new table is created
                    PhoenixUtil.createTable(conn, createTableStatement(table));
                    Map<String, String> tableParameterMap = table.getParameters();
                    tableParameterMap.put(EXTERNAL_TABLE_PURGE, "TRUE");
                    table.setParameters(tableParameterMap);
                }
            } else if (TableType.MANAGED_TABLE.name().equals(tableType)) {
                throw new MetaException("Only external table are supported for PhoenixStorageHandler");

            } else {
                throw new MetaException("Unsupported table Type: " + table.getTableType());
            }

            if (LOG.isDebugEnabled()) {
                LOG.debug("Phoenix table " + tableName + " was created");
            }
        } catch (SQLException e) {
            LOG.warn("error while creating table", e);
            throw new MetaException(e.getMessage());
        }
    }

    private String createTableStatement(Table table) throws MetaException {
        Map<String, String> tableParameterMap = table.getParameters();

        String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);
        StringBuilder ddl = new StringBuilder("create table ").append(tableName).append(" (\n");

        String phoenixRowKeys = tableParameterMap.get(PhoenixStorageHandlerConstants
                .PHOENIX_ROWKEYS);
        StringBuilder realRowKeys = new StringBuilder();
        List<String> phoenixRowKeyList = new ArrayList<>();
        for (String key:phoenixRowKeys.split(PhoenixStorageHandlerConstants.COMMA)) {
            phoenixRowKeyList.add(key.trim());
        }
        Map<String, String> columnMappingMap = getColumnMappingMap(tableParameterMap.get
                (PhoenixStorageHandlerConstants.PHOENIX_COLUMN_MAPPING));

        List<FieldSchema> fieldSchemaList = table.getSd().getCols();
        for (int i = 0, limit = fieldSchemaList.size(); i < limit; i++) {
            FieldSchema fieldSchema = fieldSchemaList.get(i);
            String fieldName = fieldSchema.getName();
            String fieldType = fieldSchema.getType();
            String columnType = PhoenixUtil.getPhoenixType(fieldType);

            String rowKeyName = getRowKeyMapping(fieldName, phoenixRowKeyList);
            if (rowKeyName != null) {
                String columnName = columnMappingMap.get(fieldName);
                if(columnName != null) {
                    rowKeyName = columnName;
                }
                // In case of RowKey
                if ("binary".equals(columnType)) {
                    // Phoenix must define max length of binary when type definition. Obtaining
                    // information from the column mapping. ex) phoenix.rowkeys = "r1, r2(100), ..."
                    List<String> tokenList =
                       new ArrayList<>();
                    for (String name: rowKeyName.split("\\(|\\)")) {
                        tokenList.add(name.trim());
                    }
                    columnType = columnType + "(" + tokenList.get(1) + ")";
                    rowKeyName = tokenList.get(0);
                }

                ddl.append("  ").append("\"").append(rowKeyName).append("\"").append(" ").append(columnType).append(" not " +
                        "null,\n");
                realRowKeys.append("\"").append(rowKeyName).append("\",");
            } else {
                // In case of Column
                String columnName = columnMappingMap.get(fieldName);

                if (columnName == null) {
                    // Use field definition.
                    columnName = fieldName;
                }

                if ("binary".equals(columnType)) {
                    // Phoenix must define max length of binary when type definition. Obtaining
                    // information from the column mapping. ex) phoenix.column.mapping=c1:c1(100)
                    List<String> tokenList = new ArrayList<>();
                    for(String name: columnName.split("\\(|\\)")){
                        tokenList.add(name.trim());
                    }
                    columnType = columnType + "(" + tokenList.get(1) + ")";
                    columnName = tokenList.get(0);
                }

                ddl.append("  ").append("\"").append(columnName).append("\"").append(" ").append(columnType).append(",\n");
            }
        }
        ddl.append("  ").append("constraint pk_").append(PhoenixUtil.getTableSchema(tableName.toUpperCase())[1]).append(" primary key(")
                .append(realRowKeys.deleteCharAt(realRowKeys.length() - 1)).append(")\n)\n");

        String tableOptions = tableParameterMap.get(PhoenixStorageHandlerConstants
                .PHOENIX_TABLE_OPTIONS);
        if (tableOptions != null) {
            ddl.append(tableOptions);
        }

        String statement = ddl.toString();

        if (LOG.isDebugEnabled()) {
            LOG.debug("DDL : " + statement);
        }

        return statement;
    }

    private String getRowKeyMapping(String rowKeyName, List<String> phoenixRowKeyList) {
        String rowKeyMapping = null;

        for (String phoenixRowKey : phoenixRowKeyList) {
            if (phoenixRowKey.equals(rowKeyName)) {
                rowKeyMapping = phoenixRowKey;
                break;
            } else if (phoenixRowKey.startsWith(rowKeyName + "(") && phoenixRowKey.endsWith(")")) {
                rowKeyMapping = phoenixRowKey;
                break;
            }
        }

        return rowKeyMapping;
    }

    @Override
    public void rollbackCreateTable(Table table) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Rollback for table : " + table.getTableName());
        }

        dropTableIfExist(table);
    }

    @Override
    public void commitCreateTable(Table table) throws MetaException {

    }

    @Override
    public void preDropTable(Table table) throws MetaException {
    }

    @Override
    public void rollbackDropTable(Table table) throws MetaException {
    }

    @Override
    public void commitDropTable(Table table, boolean deleteData) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Commit drop table : " + table.getTableName());
        }

        dropTableIfExist(table);
    }

    private void dropTableIfExist(Table table) throws MetaException {
        try (Connection conn = PhoenixConnectionUtil.getConnection(table)) {
            String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);

            if (isExternalTablePurge(table)) {
                // Drop if phoenix table exist.
                if (PhoenixUtil.existTable(conn, tableName)) {
                    PhoenixUtil.dropTable(conn, tableName);
                }
            }
        } catch (SQLException e) {
            LOG.warn("error while dropping table", e);
            throw new MetaException(e.getMessage());
        }
    }

    private  boolean isExternalTablePurge(Table table) {
        if (table == null) {
            return false;
        }
        Map<String, String> params = table.getParameters();
        if (params == null) {
            return false;
        }
        return isPropertyTrue(params, EXTERNAL_TABLE_PURGE);
    }

    private boolean isPropertyTrue(Map<String, String> tableParams, String prop) {
        return "TRUE".equalsIgnoreCase(tableParams.get(prop));
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



phoenix5-hive4/src/main/java/org/apache/phoenix/hive/PhoenixMetaHook.java [42:244]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class PhoenixMetaHook implements HiveMetaHook {

    private static final Logger LOG = LoggerFactory.getLogger(PhoenixMetaHook.class);
    private static final String EXTERNAL_TABLE_PURGE = "external.table.purge";

    @Override
    public void preCreateTable(Table table) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Precreate table : " + table.getTableName());
        }

        try (Connection conn = PhoenixConnectionUtil.getConnection(table)) {
            String tableType = table.getTableType();
            String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);

            if (TableType.EXTERNAL_TABLE.name().equals(tableType)) {
                // Check whether phoenix table exists.
                if (!PhoenixUtil.existTable(conn, tableName)) {
                    //For Hive 3.0.0 only external tables are supported. If table doesn't exist,
                    // a new table is created
                    PhoenixUtil.createTable(conn, createTableStatement(table));
                    Map<String, String> tableParameterMap = table.getParameters();
                    tableParameterMap.put(EXTERNAL_TABLE_PURGE, "TRUE");
                    table.setParameters(tableParameterMap);
                }
            } else if (TableType.MANAGED_TABLE.name().equals(tableType)) {
                throw new MetaException("Only external table are supported for PhoenixStorageHandler");

            } else {
                throw new MetaException("Unsupported table Type: " + table.getTableType());
            }

            if (LOG.isDebugEnabled()) {
                LOG.debug("Phoenix table " + tableName + " was created");
            }
        } catch (SQLException e) {
            LOG.warn("error while creating table", e);
            throw new MetaException(e.getMessage());
        }
    }

    private String createTableStatement(Table table) throws MetaException {
        Map<String, String> tableParameterMap = table.getParameters();

        String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);
        StringBuilder ddl = new StringBuilder("create table ").append(tableName).append(" (\n");

        String phoenixRowKeys = tableParameterMap.get(PhoenixStorageHandlerConstants
                .PHOENIX_ROWKEYS);
        StringBuilder realRowKeys = new StringBuilder();
        List<String> phoenixRowKeyList = new ArrayList<>();
        for (String key:phoenixRowKeys.split(PhoenixStorageHandlerConstants.COMMA)) {
            phoenixRowKeyList.add(key.trim());
        }
        Map<String, String> columnMappingMap = getColumnMappingMap(tableParameterMap.get
                (PhoenixStorageHandlerConstants.PHOENIX_COLUMN_MAPPING));

        List<FieldSchema> fieldSchemaList = table.getSd().getCols();
        for (int i = 0, limit = fieldSchemaList.size(); i < limit; i++) {
            FieldSchema fieldSchema = fieldSchemaList.get(i);
            String fieldName = fieldSchema.getName();
            String fieldType = fieldSchema.getType();
            String columnType = PhoenixUtil.getPhoenixType(fieldType);

            String rowKeyName = getRowKeyMapping(fieldName, phoenixRowKeyList);
            if (rowKeyName != null) {
                String columnName = columnMappingMap.get(fieldName);
                if(columnName != null) {
                    rowKeyName = columnName;
                }
                // In case of RowKey
                if ("binary".equals(columnType)) {
                    // Phoenix must define max length of binary when type definition. Obtaining
                    // information from the column mapping. ex) phoenix.rowkeys = "r1, r2(100), ..."
                    List<String> tokenList =
                       new ArrayList<>();
                    for (String name: rowKeyName.split("\\(|\\)")) {
                        tokenList.add(name.trim());
                    }
                    columnType = columnType + "(" + tokenList.get(1) + ")";
                    rowKeyName = tokenList.get(0);
                }

                ddl.append("  ").append("\"").append(rowKeyName).append("\"").append(" ").append(columnType).append(" not " +
                        "null,\n");
                realRowKeys.append("\"").append(rowKeyName).append("\",");
            } else {
                // In case of Column
                String columnName = columnMappingMap.get(fieldName);

                if (columnName == null) {
                    // Use field definition.
                    columnName = fieldName;
                }

                if ("binary".equals(columnType)) {
                    // Phoenix must define max length of binary when type definition. Obtaining
                    // information from the column mapping. ex) phoenix.column.mapping=c1:c1(100)
                    List<String> tokenList = new ArrayList<>();
                    for(String name: columnName.split("\\(|\\)")){
                        tokenList.add(name.trim());
                    }
                    columnType = columnType + "(" + tokenList.get(1) + ")";
                    columnName = tokenList.get(0);
                }

                ddl.append("  ").append("\"").append(columnName).append("\"").append(" ").append(columnType).append(",\n");
            }
        }
        ddl.append("  ").append("constraint pk_").append(PhoenixUtil.getTableSchema(tableName.toUpperCase())[1]).append(" primary key(")
                .append(realRowKeys.deleteCharAt(realRowKeys.length() - 1)).append(")\n)\n");

        String tableOptions = tableParameterMap.get(PhoenixStorageHandlerConstants
                .PHOENIX_TABLE_OPTIONS);
        if (tableOptions != null) {
            ddl.append(tableOptions);
        }

        String statement = ddl.toString();

        if (LOG.isDebugEnabled()) {
            LOG.debug("DDL : " + statement);
        }

        return statement;
    }

    private String getRowKeyMapping(String rowKeyName, List<String> phoenixRowKeyList) {
        String rowKeyMapping = null;

        for (String phoenixRowKey : phoenixRowKeyList) {
            if (phoenixRowKey.equals(rowKeyName)) {
                rowKeyMapping = phoenixRowKey;
                break;
            } else if (phoenixRowKey.startsWith(rowKeyName + "(") && phoenixRowKey.endsWith(")")) {
                rowKeyMapping = phoenixRowKey;
                break;
            }
        }

        return rowKeyMapping;
    }

    @Override
    public void rollbackCreateTable(Table table) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Rollback for table : " + table.getTableName());
        }

        dropTableIfExist(table);
    }

    @Override
    public void commitCreateTable(Table table) throws MetaException {

    }

    @Override
    public void preDropTable(Table table) throws MetaException {
    }

    @Override
    public void rollbackDropTable(Table table) throws MetaException {
    }

    @Override
    public void commitDropTable(Table table, boolean deleteData) throws MetaException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Commit drop table : " + table.getTableName());
        }

        dropTableIfExist(table);
    }

    private void dropTableIfExist(Table table) throws MetaException {
        try (Connection conn = PhoenixConnectionUtil.getConnection(table)) {
            String tableName = PhoenixStorageHandlerUtil.getTargetTableName(table);

            if (isExternalTablePurge(table)) {
                // Drop if phoenix table exist.
                if (PhoenixUtil.existTable(conn, tableName)) {
                    PhoenixUtil.dropTable(conn, tableName);
                }
            }
        } catch (SQLException e) {
            LOG.warn("error while dropping table", e);
            throw new MetaException(e.getMessage());
        }
    }

    private  boolean isExternalTablePurge(Table table) {
        if (table == null) {
            return false;
        }
        Map<String, String> params = table.getParameters();
        if (params == null) {
            return false;
        }
        return isPropertyTrue(params, EXTERNAL_TABLE_PURGE);
    }

    private boolean isPropertyTrue(Map<String, String> tableParams, String prop) {
        return "TRUE".equalsIgnoreCase(tableParams.get(prop));
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -