CreateTableStmt.java

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed 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 com.cloudera.impala.analysis;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.fs.permission.FsAction;

import com.cloudera.impala.authorization.Privilege;
import com.cloudera.impala.catalog.Column;
import com.cloudera.impala.catalog.HdfsStorageDescriptor;
import com.cloudera.impala.catalog.HdfsTable;
import com.cloudera.impala.catalog.RowFormat;
import com.cloudera.impala.catalog.TableLoadingException;
import com.cloudera.impala.common.AnalysisException;
import com.cloudera.impala.thrift.TAccessEvent;
import com.cloudera.impala.thrift.TCatalogObjectType;
import com.cloudera.impala.thrift.TCreateTableParams;
import com.cloudera.impala.thrift.THdfsFileFormat;
import com.cloudera.impala.thrift.TTableName;
import com.cloudera.impala.util.AvroSchemaParser;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

public class CreateTableStmt extends StatementBase {
  private final ArrayList<ColumnDef> columnDefs_;
  private final String comment_;
  private final boolean isExternal_;
  private final boolean ifNotExists_;
  private final THdfsFileFormat fileFormat_;
  private final ArrayList<ColumnDef> partitionColDefs_;
  private final RowFormat rowFormat_;
  private final TableName tableName_;
  private final Map<String, String> tblProperties_;
  private final Map<String, String> serdeProperties_;
  private final HdfsCachingOp cachingOp_;
  private HdfsUri location_;

  // Set during analysis
  private String dbName_;
  private String owner_;

  public CreateTableStmt(TableName tableName, List<ColumnDef> columnDefs,
      List<ColumnDef> partitionColumnDefs, boolean isExternal, String comment,
      RowFormat rowFormat, THdfsFileFormat fileFormat, HdfsUri location,
      HdfsCachingOp cachingOp, boolean ifNotExists, Map<String, String> tblProperties,
      Map<String, String> serdeProperties) {
    Preconditions.checkNotNull(columnDefs);
    Preconditions.checkNotNull(partitionColumnDefs);
    Preconditions.checkNotNull(fileFormat);
    Preconditions.checkNotNull(rowFormat);
    Preconditions.checkNotNull(tableName);

    this.columnDefs_ = Lists.newArrayList(columnDefs);
    this.comment_ = comment;
    this.isExternal_ = isExternal;
    this.ifNotExists_ = ifNotExists;
    this.fileFormat_ = fileFormat;
    this.location_ = location;
    this.cachingOp_ = cachingOp;
    this.partitionColDefs_ = Lists.newArrayList(partitionColumnDefs);
    this.rowFormat_ = rowFormat;
    this.tableName_ = tableName;
    this.tblProperties_ = tblProperties;
    this.serdeProperties_ = serdeProperties;
    unescapeProperties(tblProperties_);
    unescapeProperties(serdeProperties_);
  }

  public CreateTableStmt(CreateTableStmt other) {
    columnDefs_ = Lists.newArrayList(other.columnDefs_);
    comment_ = other.comment_;
    isExternal_ = other.isExternal_;
    ifNotExists_ = other.ifNotExists_;
    fileFormat_ = other.fileFormat_;
    location_ = other.location_;
    cachingOp_ = other.cachingOp_;
    partitionColDefs_ = Lists.newArrayList(other.partitionColDefs_);
    rowFormat_ = other.rowFormat_;
    tableName_ = other.tableName_;
    tblProperties_ = other.tblProperties_;
    serdeProperties_ = other.serdeProperties_;
  }

  @Override
  public CreateTableStmt clone() { return new CreateTableStmt(this); }

  public String getTbl() { return tableName_.getTbl(); }
  public TableName getTblName() { return tableName_; }
  public List<ColumnDef> getColumnDefs() { return columnDefs_; }
  public List<ColumnDef> getPartitionColumnDefs() { return partitionColDefs_; }
  public String getComment() { return comment_; }
  public boolean isExternal() { return isExternal_; }
  public boolean getIfNotExists() { return ifNotExists_; }
  public HdfsUri getLocation() { return location_; }
  public void setLocation(HdfsUri location) { this.location_ = location; }
  public THdfsFileFormat getFileFormat() { return fileFormat_; }
  public RowFormat getRowFormat() { return rowFormat_; }
  public Map<String, String> getTblProperties() { return tblProperties_; }
  public Map<String, String> getSerdeProperties() { return serdeProperties_; }

  public String getOwner() {
    Preconditions.checkNotNull(owner_);
    return owner_;
  }

  public String getDb() {
    Preconditions.checkNotNull(dbName_);
    return dbName_;
  }

  @Override
  public String toSql() {
    return ToSqlUtils.getCreateTableSql(this);
  }

  public TCreateTableParams toThrift() {
    TCreateTableParams params = new TCreateTableParams();
    params.setTable_name(new TTableName(getDb(), getTbl()));
    for (ColumnDef col: getColumnDefs()) {
      params.addToColumns(col.toThrift());
    }
    for (ColumnDef col: getPartitionColumnDefs()) {
      params.addToPartition_columns(col.toThrift());
    }
    params.setOwner(getOwner());
    params.setIs_external(isExternal());
    params.setComment(comment_);
    params.setLocation(location_ == null ? null : location_.toString());
    if (cachingOp_ != null) params.setCache_op(cachingOp_.toThrift());
    params.setRow_format(rowFormat_.toThrift());
    params.setFile_format(fileFormat_);
    params.setIf_not_exists(getIfNotExists());
    if (tblProperties_ != null) params.setTable_properties(tblProperties_);
    if (serdeProperties_ != null) params.setSerde_properties(serdeProperties_);
    return params;
  }

  @Override
  public void analyze(Analyzer analyzer) throws AnalysisException {
    Preconditions.checkState(tableName_ != null && !tableName_.isEmpty());
    tableName_.analyze();
    dbName_ = analyzer.getTargetDbName(tableName_);
    owner_ = analyzer.getUser().getName();

    if (analyzer.dbContainsTable(dbName_, tableName_.getTbl(), Privilege.CREATE) &&
        !ifNotExists_) {
      throw new AnalysisException(Analyzer.TBL_ALREADY_EXISTS_ERROR_MSG +
          String.format("%s.%s", dbName_, getTbl()));
    }

    analyzer.addAccessEvent(new TAccessEvent(dbName_ + "." + tableName_.getTbl(),
        TCatalogObjectType.TABLE, Privilege.CREATE.toString()));

    // Only Avro tables can have empty column defs because they can infer them from
    // the Avro schema.
    if (columnDefs_.isEmpty() && fileFormat_ != THdfsFileFormat.AVRO) {
      throw new AnalysisException("Table requires at least 1 column");
    }

    if (location_ != null) {
      location_.analyze(analyzer, Privilege.ALL, FsAction.READ_WRITE);
    }

    analyzeRowFormatValue(rowFormat_.getFieldDelimiter());
    analyzeRowFormatValue(rowFormat_.getLineDelimiter());
    analyzeRowFormatValue(rowFormat_.getEscapeChar());

    // Check that all the column names are valid and unique.
    analyzeColumnDefs(analyzer);

    if (fileFormat_ == THdfsFileFormat.AVRO) {
      List<ColumnDef> newColumnDefs = analyzeAvroSchema(analyzer);
      if (newColumnDefs != columnDefs_) {
        // Replace the old column defs with the new ones and analyze them.
        columnDefs_.clear();
        columnDefs_.addAll(newColumnDefs);
        analyzeColumnDefs(analyzer);
      }
    }

    if (cachingOp_ != null) cachingOp_.analyze(analyzer);
  }

  private void analyzeColumnDefs(Analyzer analyzer) throws AnalysisException {
    Set<String> colNames = Sets.newHashSet();
    for (ColumnDef colDef: columnDefs_) {
      colDef.analyze();
      if (!colNames.add(colDef.getColName().toLowerCase())) {
        throw new AnalysisException("Duplicate column name: " + colDef.getColName());
      }
    }
    for (ColumnDef colDef: partitionColDefs_) {
      colDef.analyze();
      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());
      }
    }
  }

  private List<ColumnDef> analyzeAvroSchema(Analyzer analyzer)
      throws AnalysisException {
    Preconditions.checkState(fileFormat_ == THdfsFileFormat.AVRO);
    // Look for the schema in TBLPROPERTIES and in SERDEPROPERTIES, with the latter
    // taking precedence.
    List<Map<String, String>> schemaSearchLocations = Lists.newArrayList();
    String fullTblName = dbName_ + "." + tableName_.getTbl();
    schemaSearchLocations.add(serdeProperties_);
    schemaSearchLocations.add(tblProperties_);
    String avroSchema = null;
    try {
      avroSchema = HdfsTable.getAvroSchema(schemaSearchLocations,
          dbName_ + "." + tableName_.getTbl());
    } catch (TableLoadingException e) {
      throw new AnalysisException(e.getMessage(), e);
    }

    if (Strings.isNullOrEmpty(avroSchema)) {
      throw new AnalysisException("Avro schema is null or empty: " + fullTblName);
    }

    // List of columns parsed from the Avro schema.
    List<Column> avroColumns = null;
    try {
      avroColumns = AvroSchemaParser.parse(avroSchema);
    } catch (Exception e) {
      throw new AnalysisException(String.format(
          "Error parsing Avro schema for table '%s': %s", fullTblName,
          e.getMessage()));
    }
    Preconditions.checkNotNull(avroColumns);

    // Analyze the Avro schema to detect inconsistencies with the columnDefs_.
    // In case of inconsistencies, the column defs are ignored in favor of the Avro
    // schema for simplicity and, in particular, to enable COMPUTE STATS (IMPALA-1104).
    String warnStr = null; // set if inconsistency detected
    if (avroColumns.size() != columnDefs_.size() && !columnDefs_.isEmpty()) {
      warnStr = String.format(
          "Ignoring column definitions in favor of Avro schema.\n" +
          "The Avro schema has %s column(s) but %s column definition(s) were given.",
           avroColumns.size(), columnDefs_.size());
    } else {
      // Determine whether the column names and the types match.
      for (int i = 0; i < columnDefs_.size(); ++i) {
        ColumnDef colDesc = columnDefs_.get(i);
        Column avroCol = avroColumns.get(i);
        String warnDetail = null;
        if (!colDesc.getColName().equalsIgnoreCase(avroCol.getName())) {
          warnDetail = "name";
        }
        if (colDesc.getType().isStringType() &&
            avroCol.getType().isStringType()) {
          // This is OK -- avro types for CHAR, VARCHAR, and STRING are "string"
        } else if (!colDesc.getType().equals(avroCol.getType())) {
          warnDetail = "type";
        }
        if (warnDetail != null) {
          warnStr = String.format(
              "Ignoring column definitions in favor of Avro schema due to a " +
              "mismatched column %s at position %s.\n" +
              "Column definition: %s\n" +
              "Avro schema column: %s", warnDetail, i + 1,
              colDesc.getColName() + " " + colDesc.getType().toSql(),
              avroCol.getName() + " " + avroCol.getType().toSql());
          break;
        }
      }
    }

    if (warnStr != null || columnDefs_.isEmpty()) {
      analyzer.addWarning(warnStr);
      // Create new columnDefs_ based on the Avro schema and return them.
      List<ColumnDef> avroSchemaColDefs =
          Lists.newArrayListWithCapacity(avroColumns.size());
      for (Column avroCol: avroColumns) {
        ColumnDef colDef =
            new ColumnDef(avroCol.getName(), null, avroCol.getComment());
        colDef.setType(avroCol.getType());
        avroSchemaColDefs.add(colDef);
      }
      return avroSchemaColDefs;
    }
    // The existing col defs are consistent with the Avro schema.
    return columnDefs_;
  }

  private void analyzeRowFormatValue(String value) throws AnalysisException {
    if (value == null) return;
    if (HdfsStorageDescriptor.parseDelim(value) == 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);
    }
  }

  public static void unescapeProperties(Map<String, String> propertyMap) {
    if (propertyMap == null) return;
    for (Map.Entry<String, String> kv : propertyMap.entrySet()) {
      propertyMap.put(kv.getKey(),
          new StringLiteral(kv.getValue()).getUnescapedValue());
    }
  }
}