phoenix-hive-base/src/main/java/org/apache/phoenix/hive/PhoenixMetaHook.java - phoenix-connectors - Git at Google

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

 import org.apache.hadoop.hive.metastore.HiveMetaHook;
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.phoenix.hive.constants.PhoenixStorageHandlerConstants;
 import org.apache.phoenix.hive.util.PhoenixConnectionUtil;
 import org.apache.phoenix.hive.util.PhoenixStorageHandlerUtil;
 import org.apache.phoenix.hive.util.PhoenixUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.*;

 import static org.apache.phoenix.hive.util.ColumnMappingUtils.getColumnMappingMap;

 /**
  * Implementation for notification methods which are invoked as part of transactions against the
  * hive metastore,allowing Phoenix metadata to be kept in sync with Hive's metastore.
  */
 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));
     }
 }
	/*
	* 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.hive;

	import org.apache.hadoop.hive.metastore.HiveMetaHook;
	import org.apache.hadoop.hive.metastore.TableType;
	import org.apache.hadoop.hive.metastore.api.FieldSchema;
	import org.apache.hadoop.hive.metastore.api.MetaException;
	import org.apache.hadoop.hive.metastore.api.Table;
	import org.apache.phoenix.hive.constants.PhoenixStorageHandlerConstants;
	import org.apache.phoenix.hive.util.PhoenixConnectionUtil;
	import org.apache.phoenix.hive.util.PhoenixStorageHandlerUtil;
	import org.apache.phoenix.hive.util.PhoenixUtil;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.sql.Connection;
	import java.sql.SQLException;
	import java.util.*;

	import static org.apache.phoenix.hive.util.ColumnMappingUtils.getColumnMappingMap;

	/**
	* Implementation for notification methods which are invoked as part of transactions against the
	* hive metastore,allowing Phoenix metadata to be kept in sync with Hive's metastore.
	*/
	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));
	}
	}