/**
 * Copyright 2013 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 org.kitesdk.cli.commands;

import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.util.List;
import org.kitesdk.data.Dataset;
import org.kitesdk.data.DatasetDescriptor;
import org.kitesdk.data.Datasets;
import org.kitesdk.data.Formats;
import org.slf4j.Logger;

@Parameters(commandDescription = "Create an empty dataset")
public class CreateDatasetCommand extends BaseDatasetCommand {

  @Parameter(description = "<dataset name>")
  List<String> datasets;

  @Parameter(names = {"-s", "--schema"}, required=true,
      description = "The file containing the Avro schema.")
  String avroSchemaFile;

  @Parameter(names = {"-p", "--partition-by"},
      description = "The file containing a JSON-formatted partition strategy.")
  String partitionStrategyFile;

  @Parameter(names = {"-f", "--format"},
      description = "The file format: avro or parquet.")
  String format = Formats.AVRO.getName();

  public CreateDatasetCommand(Logger console) {
    super(console);
  }

  @Override
  public int run() throws IOException {
    if (datasets == null || datasets.size() != 1) {
      throw new IllegalArgumentException(
          "Exactly one dataset name must be specified.");
    }

    DatasetDescriptor.Builder descriptorBuilder = new DatasetDescriptor.Builder();

    if (format.equals(Formats.AVRO.getName())) {
      descriptorBuilder.format(Formats.AVRO);
    } else if (format.equals(Formats.PARQUET.getName())) {
      descriptorBuilder.format(Formats.PARQUET);
    } else {
      throw new IllegalArgumentException("Unrecognized format: " + format);
    }

    descriptorBuilder.schemaUri(qualifiedURI(avroSchemaFile));

    if (partitionStrategyFile != null) {
      descriptorBuilder.partitionStrategy(open(partitionStrategyFile));
    }

    DatasetDescriptor descriptor = descriptorBuilder.build();
    if (isDataUri(datasets.get(0))) {
      Datasets.<Object, Dataset<Object>> create(datasets.get(0), descriptor);
    } else {
      getDatasetRepository().create(datasets.get(0), descriptor);
    }
    console.debug("Created dataset {}", datasets.get(0));

    return 0;
  }

  @Override
  public List<String> getExamples() {
    return Lists.newArrayList(
        "# Create dataset \"users\" in Hive:",
        "users --schema user.avsc",
        "# Create dataset \"users\" using parquet:",
        "users --schema user.avsc --format parquet",
        "# Create dataset \"users\" partitioned by JSON configuration:",
        "users --schema user.avsc --partition-by user_part.json"
    );
  }

}

@@ -1454,24 +1454,171 @@
 ategyFile;%0A%0A
+  @Parameter(names = %7B%22-m%22, %22--mapping%22%7D,%0A      description = %22The file containing a JSON-formatted column mapping.%22)%0A  String columnMappingFile;%0A%0A
   @Parameter
@@ -2563,13 +2563,24 @@
 tegy
-(open
+Uri(qualifiedURI
 (par
@@ -2600,32 +2600,151 @@
 yFile));%0A    %7D%0A%0A
+    if (columnMappingFile != null) %7B%0A      descriptorBuilder.columnMappingUri(qualifiedURI(columnMappingFile));%0A    %7D%0A%0A
     DatasetDescr
@@ -3017,24 +3017,16 @@
 Created 
-dataset 
 %7B%7D%22, dat

Add column mapping option to create command.