| // 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.impala.analysis; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.LinkedHashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.stream.Collectors; |
| |
| import org.apache.hadoop.fs.permission.FsAction; |
| import org.apache.hadoop.hive.metastore.api.SQLForeignKey; |
| import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey; |
| import org.apache.impala.authorization.Privilege; |
| import org.apache.impala.catalog.Column; |
| import org.apache.impala.catalog.FeFsTable; |
| import org.apache.impala.catalog.FeTable; |
| import org.apache.impala.catalog.HdfsStorageDescriptor; |
| import org.apache.impala.catalog.RowFormat; |
| import org.apache.impala.catalog.Type; |
| import org.apache.impala.common.AnalysisException; |
| import org.apache.impala.common.FileSystemUtil; |
| import org.apache.impala.common.Pair; |
| import org.apache.impala.thrift.TAccessEvent; |
| import org.apache.impala.thrift.TCatalogObjectType; |
| import org.apache.impala.thrift.THdfsFileFormat; |
| import org.apache.impala.thrift.TQueryOptions; |
| import org.apache.impala.thrift.TSortingOrder; |
| import org.apache.impala.util.AcidUtils; |
| import org.apache.impala.util.MetaStoreUtil; |
| import org.apache.thrift.TException; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.Lists; |
| |
| /** |
| * Represents the table parameters in a CREATE TABLE statement. These parameters |
| * correspond to the following clauses in a CREATE TABLE statement: |
| * - EXTERNAL |
| * - IF NOT EXISTS |
| * - PARTITIONED BY |
| * - PARTITION BY |
| * - ROWFORMAT |
| * - FILEFORMAT |
| * - COMMENT |
| * - SERDEPROPERTIES |
| * - TBLPROPERTIES |
| * - LOCATION |
| * - CACHED IN |
| * - SORT BY |
| */ |
| class TableDef { |
| // Name of the new table |
| private final TableName tableName_; |
| |
| // List of column definitions |
| private final List<ColumnDef> columnDefs_ = new ArrayList<>(); |
| |
| // Names of primary key columns. Populated by the parser. An empty value doesn't |
| // mean no primary keys were specified as the columnDefs_ could contain primary keys. |
| private final List<String> primaryKeyColNames_ = new ArrayList<>(); |
| |
| // If true, the table's data will be preserved if dropped. |
| private final boolean isExternal_; |
| |
| // If true, no errors are thrown if the table already exists. |
| private final boolean ifNotExists_; |
| |
| // Partitioning parameters. |
| private final TableDataLayout dataLayout_; |
| |
| ///////////////////////////////////////// |
| // BEGIN: Members that need to be reset() |
| |
| // Authoritative list of primary key column definitions populated during analysis. |
| private final List<ColumnDef> primaryKeyColDefs_ = new ArrayList<>(); |
| |
| // Hive primary keys and foreign keys structures populated during analysis. |
| List<SQLPrimaryKey> sqlPrimaryKeys_ = new ArrayList<>(); |
| List<SQLForeignKey> sqlForeignKeys_ = new ArrayList<>(); |
| |
| public List<SQLPrimaryKey> getSqlPrimaryKeys() { |
| return sqlPrimaryKeys_; |
| } |
| |
| public List<SQLForeignKey> getSqlForeignKeys() { |
| return sqlForeignKeys_; |
| } |
| |
| // True if analyze() has been called. |
| private boolean isAnalyzed_ = false; |
| |
| // Generated Kudu properties set during analysis. |
| private Map<String, String> generatedKuduProperties_ = new HashMap<>(); |
| |
| // END: Members that need to be reset() |
| ///////////////////////////////////////// |
| |
| /** |
| * Set of table options. These options are grouped together for convenience while |
| * parsing CREATE TABLE statements. They are typically found at the end of CREATE |
| * TABLE statements. |
| */ |
| static class Options { |
| // Optional list of columns to sort data by when inserting into this table. |
| final List<String> sortCols; |
| |
| // Comment to attach to the table |
| final String comment; |
| |
| // Custom row format of the table. Leave null to specify default row format. |
| final RowFormat rowFormat; |
| |
| // Key/values to persist with table serde metadata. |
| final Map<String, String> serdeProperties; |
| |
| // File format of the table |
| final THdfsFileFormat fileFormat; |
| |
| // The HDFS location of where the table data will stored. |
| final HdfsUri location; |
| |
| // The HDFS caching op that should be applied to this table. |
| final HdfsCachingOp cachingOp; |
| |
| // Key/values to persist with table metadata. |
| final Map<String, String> tblProperties; |
| |
| // Sorting order for SORT BY queries. |
| final TSortingOrder sortingOrder; |
| |
| Options(Pair<List<String>, TSortingOrder> sortProperties, String comment, |
| RowFormat rowFormat, Map<String, String> serdeProperties, |
| THdfsFileFormat fileFormat, HdfsUri location, HdfsCachingOp cachingOp, |
| Map<String, String> tblProperties, TQueryOptions queryOptions) { |
| this.sortCols = sortProperties.first; |
| this.sortingOrder = sortProperties.second; |
| this.comment = comment; |
| this.rowFormat = rowFormat; |
| Preconditions.checkNotNull(serdeProperties); |
| this.serdeProperties = serdeProperties; |
| // The file format passed via STORED AS <file format> has a higher precedence than |
| // the one set in query options. |
| this.fileFormat = (fileFormat != null) ? |
| fileFormat : queryOptions.getDefault_file_format(); |
| this.location = location; |
| this.cachingOp = cachingOp; |
| Preconditions.checkNotNull(tblProperties); |
| this.tblProperties = tblProperties; |
| } |
| |
| public Options(String comment, TQueryOptions queryOptions) { |
| // Passing null to file format so that it uses the file format from the query option |
| // if specified, otherwise it will use the default file format, which is TEXT. |
| this(new Pair<>(ImmutableList.of(), TSortingOrder.LEXICAL), comment, |
| RowFormat.DEFAULT_ROW_FORMAT, new HashMap<>(), /* file format */null, null, |
| null, new HashMap<>(), queryOptions); |
| } |
| } |
| |
| private Options options_; |
| |
| /** |
| * Primary Key attributes grouped together to be populated by the parser. |
| * Currently only defined for HDFS tables. |
| */ |
| static class PrimaryKey { |
| |
| // Primary key table name |
| final TableName pkTableName; |
| |
| // Primary Key columns |
| final List<String> primaryKeyColNames; |
| |
| // Primary Key constraint name |
| final String pkConstraintName; |
| |
| // Constraints |
| final boolean relyCstr; |
| final boolean validateCstr; |
| final boolean enableCstr; |
| |
| |
| public PrimaryKey(TableName pkTableName, List<String> primaryKeyColNames, |
| String pkConstraintName, boolean relyCstr, |
| boolean validateCstr, boolean enableCstr) { |
| this.pkTableName = pkTableName; |
| this.primaryKeyColNames = primaryKeyColNames; |
| this.pkConstraintName = pkConstraintName; |
| this.relyCstr = relyCstr; |
| this.validateCstr = validateCstr; |
| this.enableCstr = enableCstr; |
| } |
| |
| public TableName getPkTableName() { |
| return pkTableName; |
| } |
| |
| public List<String> getPrimaryKeyColNames() { |
| return primaryKeyColNames; |
| } |
| |
| public String getPkConstraintName() { |
| return pkConstraintName; |
| } |
| |
| public boolean isRelyCstr() { |
| return relyCstr; |
| } |
| |
| public boolean isValidateCstr() { |
| return validateCstr; |
| } |
| |
| public boolean isEnableCstr() { |
| return enableCstr; |
| } |
| } |
| |
| |
| /** |
| * Foreign Key attributes grouped together to be populated by the parser. |
| * Currently only defined for HDFS tables. An FK definition is of the form |
| * "foreign key(col1, col2) references pk_tbl(col3, col4)" |
| */ |
| static class ForeignKey { |
| // Primary key table |
| final TableName pkTableName; |
| |
| // Primary key cols |
| final List<String> primaryKeyColNames; |
| |
| // Foreign key cols |
| final List<String> foreignKeyColNames; |
| |
| // Name of fk |
| final String fkConstraintName; |
| |
| // Fully qualified pk name. Set during analysis. |
| TableName fullyQualifiedPkTableName; |
| |
| // Constraints |
| final boolean relyCstr; |
| final boolean validateCstr; |
| final boolean enableCstr; |
| |
| ForeignKey(TableName pkTableName, List<String> primaryKeyColNames, |
| List<String> foreignKeyColNames, String fkName, boolean relyCstr, |
| boolean validateCstr, boolean enableCstr) { |
| this.pkTableName = pkTableName; |
| this.primaryKeyColNames = primaryKeyColNames; |
| this.foreignKeyColNames = foreignKeyColNames; |
| this.relyCstr = relyCstr; |
| this.validateCstr = validateCstr; |
| this.enableCstr = enableCstr; |
| this.fkConstraintName = fkName; |
| } |
| |
| public TableName getPkTableName() { |
| return pkTableName; |
| } |
| |
| public List<String> getPrimaryKeyColNames() { |
| return primaryKeyColNames; |
| } |
| |
| public List<String> getForeignKeyColNames() { |
| return foreignKeyColNames; |
| } |
| |
| public String getFkConstraintName() { |
| return fkConstraintName; |
| } |
| |
| public TableName getFullyQualifiedPkTableName() { |
| return fullyQualifiedPkTableName; |
| } |
| |
| public boolean isRelyCstr() { |
| return relyCstr; |
| } |
| |
| public boolean isValidateCstr() { |
| return validateCstr; |
| } |
| |
| public boolean isEnableCstr() { |
| return enableCstr; |
| } |
| } |
| |
| // A TableDef will have only one primary key. |
| private PrimaryKey primaryKey_; |
| |
| // There maybe multiple foreign keys for a TableDef forming multiple PK-FK |
| // relationships. |
| private List<ForeignKey> foreignKeysList_ = new ArrayList<>(); |
| |
| // Result of analysis. |
| private TableName fqTableName_; |
| |
| TableDef(TableName tableName, boolean isExternal, boolean ifNotExists) { |
| tableName_ = tableName; |
| isExternal_ = isExternal; |
| ifNotExists_ = ifNotExists; |
| dataLayout_ = TableDataLayout.createEmptyLayout(); |
| } |
| |
| public void reset() { |
| primaryKeyColDefs_.clear(); |
| columnDefs_.clear(); |
| isAnalyzed_ = false; |
| generatedKuduProperties_.clear(); |
| } |
| |
| public TableName getTblName() { |
| return fqTableName_ != null ? fqTableName_ : tableName_; |
| } |
| public String getTbl() { return tableName_.getTbl(); } |
| public boolean isAnalyzed() { return isAnalyzed_; } |
| List<ColumnDef> getColumnDefs() { return columnDefs_; } |
| List<String> getColumnNames() { return ColumnDef.toColumnNames(columnDefs_); } |
| |
| List<Type> getColumnTypes() { |
| return columnDefs_.stream().map(col -> col.getType()).collect(Collectors.toList()); |
| } |
| |
| public void setPrimaryKey(TableDef.PrimaryKey primaryKey_) { |
| this.primaryKey_ = primaryKey_; |
| } |
| List<String> getPartitionColumnNames() { |
| return ColumnDef.toColumnNames(getPartitionColumnDefs()); |
| } |
| |
| List<ColumnDef> getPartitionColumnDefs() { |
| return dataLayout_.getPartitionColumnDefs(); |
| } |
| |
| boolean isKuduTable() { return options_.fileFormat == THdfsFileFormat.KUDU; } |
| List<String> getPrimaryKeyColumnNames() { return primaryKeyColNames_; } |
| List<ColumnDef> getPrimaryKeyColumnDefs() { return primaryKeyColDefs_; } |
| boolean isExternal() { return isExternal_; } |
| boolean getIfNotExists() { return ifNotExists_; } |
| Map<String, String> getGeneratedKuduProperties() { return generatedKuduProperties_; } |
| void putGeneratedKuduProperty(String key, String value) { |
| Preconditions.checkNotNull(key); |
| generatedKuduProperties_.put(key, value); |
| } |
| List<KuduPartitionParam> getKuduPartitionParams() { |
| return dataLayout_.getKuduPartitionParams(); |
| } |
| void setOptions(Options options) { |
| Preconditions.checkNotNull(options); |
| options_ = options; |
| } |
| List<String> getSortColumns() { return options_.sortCols; } |
| String getComment() { return options_.comment; } |
| Map<String, String> getTblProperties() { return options_.tblProperties; } |
| HdfsCachingOp getCachingOp() { return options_.cachingOp; } |
| HdfsUri getLocation() { return options_.location; } |
| Map<String, String> getSerdeProperties() { return options_.serdeProperties; } |
| THdfsFileFormat getFileFormat() { return options_.fileFormat; } |
| RowFormat getRowFormat() { return options_.rowFormat; } |
| TSortingOrder getSortingOrder() { return options_.sortingOrder; } |
| List<ForeignKey> getForeignKeysList() { return foreignKeysList_; } |
| |
| /** |
| * Analyzes the parameters of a CREATE TABLE statement. |
| */ |
| void analyze(Analyzer analyzer) throws AnalysisException { |
| if (isAnalyzed_) return; |
| Preconditions.checkState(tableName_ != null && !tableName_.isEmpty()); |
| fqTableName_ = analyzer.getFqTableName(getTblName()); |
| fqTableName_.analyze(); |
| analyzeAcidProperties(analyzer); |
| analyzeColumnDefs(analyzer); |
| analyzePrimaryKeys(); |
| analyzeForeignKeys(analyzer); |
| |
| if (analyzer.dbContainsTable(getTblName().getDb(), getTbl(), Privilege.CREATE) |
| && !getIfNotExists()) { |
| throw new AnalysisException(Analyzer.TBL_ALREADY_EXISTS_ERROR_MSG + getTblName()); |
| } |
| |
| analyzer.addAccessEvent(new TAccessEvent(fqTableName_.toString(), |
| TCatalogObjectType.TABLE, Privilege.CREATE.toString())); |
| |
| Preconditions.checkNotNull(options_); |
| analyzeOptions(analyzer); |
| isAnalyzed_ = true; |
| } |
| |
| /** |
| * Analyzes table and partition column definitions, checking whether all column |
| * names are unique. |
| */ |
| private void analyzeColumnDefs(Analyzer analyzer) throws AnalysisException { |
| Set<String> colNames = new HashSet<>(); |
| for (ColumnDef colDef: columnDefs_) { |
| colDef.analyze(analyzer); |
| if (!colNames.add(colDef.getColName().toLowerCase())) { |
| throw new AnalysisException("Duplicate column name: " + colDef.getColName()); |
| } |
| if (!isKuduTable() && colDef.hasKuduOptions()) { |
| throw new AnalysisException(String.format("Unsupported column options for " + |
| "file format '%s': '%s'", getFileFormat().name(), colDef.toString())); |
| } |
| } |
| for (ColumnDef colDef: getPartitionColumnDefs()) { |
| colDef.analyze(analyzer); |
| if (!colDef.getType().supportsTablePartitioning()) { |
| throw new AnalysisException( |
| String.format("Type '%s' is not supported as partition-column type " + |
| "in column: %s", colDef.getType().toSql(), colDef.getColName())); |
| } |
| if (!colNames.add(colDef.getColName().toLowerCase())) { |
| throw new AnalysisException("Duplicate column name: " + colDef.getColName()); |
| } |
| } |
| } |
| |
| /** |
| * Analyzes the primary key columns. Checks if the specified primary key columns exist |
| * in the table column definitions and if composite primary keys are properly defined |
| * using the PRIMARY KEY (col,..col) clause. |
| */ |
| private void analyzePrimaryKeys() throws AnalysisException { |
| for (ColumnDef colDef: columnDefs_) { |
| if (colDef.isPrimaryKey()) primaryKeyColDefs_.add(colDef); |
| } |
| if (primaryKeyColDefs_.size() > 1) { |
| throw new AnalysisException("Multiple primary keys specified. " + |
| "Composite primary keys can be specified using the " + |
| "PRIMARY KEY (col1, col2, ...) syntax at the end of the column definition."); |
| } |
| |
| if (primaryKeyColNames_.isEmpty()) { |
| if (primaryKey_ == null || primaryKey_.getPrimaryKeyColNames().isEmpty()) { |
| return; |
| } else { |
| primaryKeyColNames_.addAll(primaryKey_.getPrimaryKeyColNames()); |
| } |
| } |
| |
| if (!primaryKeyColDefs_.isEmpty()) { |
| throw new AnalysisException("Multiple primary keys specified. " + |
| "Composite primary keys can be specified using the " + |
| "PRIMARY KEY (col1, col2, ...) syntax at the end of the column definition."); |
| } |
| Map<String, ColumnDef> colDefsByColName = ColumnDef.mapByColumnNames(columnDefs_); |
| int keySeq = 1; |
| String constraintName = null; |
| for (String colName: primaryKeyColNames_) { |
| colName = colName.toLowerCase(); |
| ColumnDef colDef = colDefsByColName.remove(colName); |
| if (colDef == null) { |
| if (ColumnDef.toColumnNames(primaryKeyColDefs_).contains(colName)) { |
| throw new AnalysisException(String.format("Column '%s' is listed multiple " + |
| "times as a PRIMARY KEY.", colName)); |
| } |
| throw new AnalysisException(String.format( |
| "PRIMARY KEY column '%s' does not exist in the table", colName)); |
| } |
| if (colDef.isExplicitNullable()) { |
| throw new AnalysisException("Primary key columns cannot be nullable: " + |
| colDef.toString()); |
| } |
| // HDFS Table specific analysis. |
| if (primaryKey_ != null) { |
| // We do not support enable and validate for primary keys. |
| if (primaryKey_.isEnableCstr()) { |
| throw new AnalysisException("ENABLE feature is not supported yet."); |
| } |
| if (primaryKey_.isValidateCstr()) { |
| throw new AnalysisException("VALIDATE feature is not supported yet."); |
| } |
| // All primary keys in a composite key should have the same constraint name. This |
| // is necessary because of HIVE-16603. See IMPALA-9188 for details. |
| if (constraintName == null) { |
| constraintName = generateConstraintName(); |
| } |
| // Each column of a primary key definition will be an SQLPrimaryKey. |
| sqlPrimaryKeys_.add(new SQLPrimaryKey(getTblName().getDb(), getTbl(), |
| colDef.getColName(), keySeq++, constraintName, primaryKey_.enableCstr, |
| primaryKey_.validateCstr, primaryKey_.relyCstr)); |
| } |
| primaryKeyColDefs_.add(colDef); |
| } |
| } |
| |
| private void analyzeForeignKeys(Analyzer analyzer) throws AnalysisException { |
| if (foreignKeysList_ == null || foreignKeysList_.size() == 0) return; |
| for (ForeignKey fk: foreignKeysList_) { |
| // Foreign Key and Primary Key columns don't match. |
| if (fk.getForeignKeyColNames().size() != fk.getPrimaryKeyColNames().size()){ |
| throw new AnalysisException("The number of foreign key columns should be same" + |
| " as the number of parent key columns."); |
| } |
| String parentDb = fk.getPkTableName().getDb(); |
| if (parentDb == null) { |
| parentDb = analyzer.getDefaultDb(); |
| } |
| fk.fullyQualifiedPkTableName = new TableName(parentDb, fk.pkTableName.getTbl()); |
| //Check if parent table exits |
| if (!analyzer.dbContainsTable(parentDb, fk.getPkTableName().getTbl(), |
| Privilege.VIEW_METADATA)) { |
| throw new AnalysisException("Parent table not found: " |
| + analyzer.getFqTableName(fk.getPkTableName())); |
| } |
| |
| //Check for primary key cols in parent table |
| FeTable parentTable = analyzer.getTable(fk.getPkTableName(), |
| Privilege.VIEW_METADATA); |
| |
| if (!(parentTable instanceof FeFsTable)) { |
| throw new AnalysisException("Foreign keys on non-HDFS parent tables are not " |
| + "supported."); |
| } |
| |
| for (String pkCol : fk.getPrimaryKeyColNames()) { |
| // TODO: Check column types of parent table and child tables match. Currently HMS |
| // API fails if they don't, it's good to fail early during analysis here. |
| if (!parentTable.getColumnNames().contains(pkCol.toLowerCase())) { |
| throw new AnalysisException("Parent column not found: " + pkCol.toLowerCase()); |
| } |
| // Hive has a bug that prevents foreign keys from being added when pk column is |
| // not part of primary key. This can be confusing. Till this bug is fixed, we |
| // will not allow foreign keys definition on such columns. |
| try { |
| if (!((FeFsTable) parentTable).getPrimaryKeyColumnNames().contains(pkCol)) { |
| throw new AnalysisException(String.format("Parent column %s is not part of " |
| + "primary key.", pkCol)); |
| } |
| } catch (TException e) { |
| // In local catalog mode, we do not aggressively load PK/FK information, a |
| // call to getPrimaryKeyColumnNames() will try to selectively load PK/FK |
| // information. Hence, TException is thrown only in local catalog mode. |
| throw new AnalysisException("Failed to get primary key columns for " |
| + fk.pkTableName); |
| } |
| } |
| |
| // We do not support ENABLE and VALIDATE. |
| if (fk.isEnableCstr()) { |
| throw new AnalysisException("ENABLE feature is not supported yet."); |
| } |
| |
| if (fk.isValidateCstr()) { |
| throw new AnalysisException("VALIDATE feature is not supported yet."); |
| } |
| |
| String constraintName = null; |
| for (int i = 0; i < fk.getForeignKeyColNames().size(); i++) { |
| if (fk.getFkConstraintName() == null) { |
| if (i == 0){ |
| constraintName = generateConstraintName(); |
| } |
| } else { |
| constraintName = fk.getFkConstraintName(); |
| } |
| SQLForeignKey sqlForeignKey = new SQLForeignKey(); |
| sqlForeignKey.setPktable_db(parentDb); |
| sqlForeignKey.setPktable_name(fk.getPkTableName().getTbl()); |
| sqlForeignKey.setFktable_db(getTblName().getDb()); |
| sqlForeignKey.setFktable_name(getTbl()); |
| sqlForeignKey.setPkcolumn_name(fk.getPrimaryKeyColNames().get(i).toLowerCase()); |
| sqlForeignKey.setFk_name(constraintName); |
| sqlForeignKey.setKey_seq(i+1); |
| sqlForeignKey.setFkcolumn_name(fk.getForeignKeyColNames().get(i).toLowerCase()); |
| sqlForeignKey.setRely_cstr(fk.isRelyCstr()); |
| getSqlForeignKeys().add(sqlForeignKey); |
| } |
| } |
| } |
| |
| /** |
| * Utility method to generate a unique constraint name when user does not specify one. |
| * TODO: Collisions possible? HMS doesn't have an API to query existing constraint |
| * names. |
| */ |
| private String generateConstraintName() { |
| return UUID.randomUUID().toString(); |
| } |
| |
| /** |
| * Analyzes the list of columns in 'sortCols' against the columns of 'table' and |
| * returns their matching positions in the table's columns. Each column of 'sortCols' |
| * must occur in 'table' as a non-partitioning column. 'table' must be an HDFS table. |
| * If there are errors during the analysis, this will throw an AnalysisException. |
| */ |
| public static List<Integer> analyzeSortColumns(List<String> sortCols, FeTable table, |
| TSortingOrder sortingOrder) throws AnalysisException { |
| Preconditions.checkState(table instanceof FeFsTable); |
| |
| List<Type> columnTypes = table.getNonClusteringColumns().stream().map( |
| col -> col.getType()).collect(Collectors.toList()); |
| return analyzeSortColumns(sortCols, |
| Column.toColumnNames(table.getNonClusteringColumns()), |
| Column.toColumnNames(table.getClusteringColumns()), columnTypes, sortingOrder); |
| } |
| |
| /** |
| * Analyzes the list of columns in 'sortCols' and returns their matching positions in |
| * 'tableCols'. Each column must occur in 'tableCols' and must not occur in |
| * 'partitionCols'. If there are errors during the analysis, this will throw an |
| * AnalysisException. |
| */ |
| public static List<Integer> analyzeSortColumns(List<String> sortCols, |
| List<String> tableCols, List<String> partitionCols, List<Type> columnTypes, |
| TSortingOrder sortingOrder) throws AnalysisException { |
| // The index of each sort column in the list of table columns. |
| Set<Integer> colIdxs = new LinkedHashSet<>(); |
| |
| int numColumns = 0; |
| for (String sortColName: sortCols) { |
| ++numColumns; |
| // Make sure it's not a partition column. |
| if (partitionCols.contains(sortColName)) { |
| throw new AnalysisException(String.format("SORT BY column list must not " + |
| "contain partition column: '%s'", sortColName)); |
| } |
| |
| // Determine the index of each sort column in the list of table columns. |
| boolean foundColumn = false; |
| for (int j = 0; j < tableCols.size(); ++j) { |
| if (tableCols.get(j).equalsIgnoreCase(sortColName)) { |
| if (colIdxs.contains(j)) { |
| throw new AnalysisException(String.format("Duplicate column in SORT BY " + |
| "list: %s", sortColName)); |
| } |
| colIdxs.add(j); |
| foundColumn = true; |
| break; |
| } |
| } |
| if (!foundColumn) { |
| throw new AnalysisException(String.format("Could not find SORT BY column " + |
| "'%s' in table.", sortColName)); |
| } |
| } |
| |
| // Analyzing Z-Order specific constraints |
| if (sortingOrder == TSortingOrder.ZORDER) { |
| if (numColumns == 1) { |
| throw new AnalysisException(String.format("SORT BY ZORDER with 1 column is " + |
| "equivalent to SORT BY. Please, use the latter, if that was your " + |
| "intention.")); |
| } |
| |
| List<? extends Type> notSupportedTypes = Arrays.asList(Type.STRING, Type.VARCHAR, |
| Type.FLOAT, Type.DOUBLE); |
| for (Integer position : colIdxs) { |
| Type colType = columnTypes.get(position); |
| |
| if (notSupportedTypes.stream().anyMatch(type -> colType.matchesType(type))) { |
| throw new AnalysisException(String.format("SORT BY ZORDER does not support " |
| + "column types: %s", String.join(", ", |
| notSupportedTypes.stream().map(type -> type.toString()) |
| .collect(Collectors.toList())))); |
| } |
| } |
| } |
| |
| Preconditions.checkState(numColumns == colIdxs.size()); |
| return Lists.newArrayList(colIdxs); |
| } |
| |
| private void analyzeOptions(Analyzer analyzer) throws AnalysisException { |
| MetaStoreUtil.checkShortPropertyMap("Property", options_.tblProperties); |
| MetaStoreUtil.checkShortPropertyMap("Serde property", options_.serdeProperties); |
| |
| if (options_.location != null) { |
| options_.location.analyze(analyzer, Privilege.ALL, FsAction.READ_WRITE); |
| } |
| |
| if (options_.cachingOp != null) { |
| options_.cachingOp.analyze(analyzer); |
| if (options_.cachingOp.shouldCache() && options_.location != null && |
| !FileSystemUtil.isPathCacheable(options_.location.getPath())) { |
| throw new AnalysisException(String.format("Location '%s' cannot be cached. " + |
| "Please retry without caching: CREATE TABLE ... UNCACHED", |
| options_.location)); |
| } |
| } |
| |
| // Analyze 'skip.header.line.format' property. |
| AlterTableSetTblProperties.analyzeSkipHeaderLineCount(options_.tblProperties); |
| analyzeRowFormat(analyzer); |
| |
| String sortByKey = AlterTableSortByStmt.TBL_PROP_SORT_COLUMNS; |
| String sortOrderKey = AlterTableSortByStmt.TBL_PROP_SORT_ORDER; |
| if (options_.tblProperties.containsKey(sortByKey)) { |
| throw new AnalysisException(String.format("Table definition must not contain the " + |
| "%s table property. Use SORT BY (...) instead.", sortByKey)); |
| } |
| |
| if (options_.tblProperties.containsKey(sortOrderKey)) { |
| throw new AnalysisException(String.format("Table definition must not contain the " + |
| "%s table property. Use SORT BY %s (...) instead.", sortOrderKey, |
| options_.sortingOrder.toString())); |
| } |
| |
| // Analyze sort columns. |
| if (options_.sortCols == null) return; |
| if (isKuduTable()) { |
| throw new AnalysisException(String.format("SORT BY is not supported for Kudu "+ |
| "tables.")); |
| } |
| |
| analyzeSortColumns(options_.sortCols, getColumnNames(), getPartitionColumnNames(), |
| getColumnTypes(), options_.sortingOrder); |
| } |
| |
| private void analyzeRowFormat(Analyzer analyzer) throws AnalysisException { |
| if (options_.rowFormat == null) return; |
| if (isKuduTable()) { |
| throw new AnalysisException(String.format( |
| "ROW FORMAT cannot be specified for file format %s.", options_.fileFormat)); |
| } |
| |
| Byte fieldDelim = analyzeRowFormatValue(options_.rowFormat.getFieldDelimiter()); |
| Byte lineDelim = analyzeRowFormatValue(options_.rowFormat.getLineDelimiter()); |
| Byte escapeChar = analyzeRowFormatValue(options_.rowFormat.getEscapeChar()); |
| if (options_.fileFormat == THdfsFileFormat.TEXT) { |
| if (fieldDelim == null) fieldDelim = HdfsStorageDescriptor.DEFAULT_FIELD_DELIM; |
| if (lineDelim == null) lineDelim = HdfsStorageDescriptor.DEFAULT_LINE_DELIM; |
| if (escapeChar == null) escapeChar = HdfsStorageDescriptor.DEFAULT_ESCAPE_CHAR; |
| if (fieldDelim.equals(lineDelim)) { |
| throw new AnalysisException("Field delimiter and line delimiter have same " + |
| "value: byte " + fieldDelim); |
| } |
| if (fieldDelim.equals(escapeChar)) { |
| analyzer.addWarning("Field delimiter and escape character have same value: " + |
| "byte " + fieldDelim + ". Escape character will be ignored"); |
| } |
| if (lineDelim.equals(escapeChar)) { |
| analyzer.addWarning("Line delimiter and escape character have same value: " + |
| "byte " + lineDelim + ". Escape character will be ignored"); |
| } |
| } |
| } |
| |
| private Byte analyzeRowFormatValue(String value) throws AnalysisException { |
| if (value == null) return null; |
| Byte byteVal = HdfsStorageDescriptor.parseDelim(value); |
| if (byteVal == null) { |
| throw new AnalysisException("ESCAPED BY values and LINE/FIELD " + |
| "terminators must be specified as a single character or as a decimal " + |
| "value in the range [-128:127]: " + value); |
| } |
| return byteVal; |
| } |
| |
| /** |
| * Analyzes Hive ACID related properties. |
| * Can change table properties based on query options. |
| */ |
| private void analyzeAcidProperties(Analyzer analyzer) throws AnalysisException { |
| if (isExternal_) { |
| if (AcidUtils.isTransactionalTable(options_.tblProperties)) { |
| throw new AnalysisException("EXTERNAL tables cannot be transactional"); |
| } |
| return; |
| } |
| |
| if (options_.fileFormat == THdfsFileFormat.KUDU) { |
| if (AcidUtils.isTransactionalTable(options_.tblProperties)) { |
| throw new AnalysisException("Kudu tables cannot be transactional"); |
| } |
| return; |
| } |
| |
| AcidUtils.setTransactionalProperties(options_.tblProperties, |
| analyzer.getQueryOptions().getDefault_transactional_type()); |
| // Disallow creation of full ACID table. |
| analyzer.ensureTableNotFullAcid(options_.tblProperties, fqTableName_.toString()); |
| } |
| } |
