package com.facebook.presto.metadata;

import com.facebook.presto.ingest.ImportSchemaUtil;
import com.facebook.presto.spi.ImportClient;
import com.facebook.presto.spi.SchemaField;
import com.facebook.presto.split.ImportClientFactory;
import com.facebook.presto.sql.tree.QualifiedName;
import com.facebook.presto.tuple.TupleInfo;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.inject.Inject;

import java.util.List;
import java.util.concurrent.Callable;

import static com.facebook.presto.util.RetryDriver.runWithRetryUnchecked;
import static com.google.common.base.Preconditions.checkNotNull;

public class ImportMetadata
        implements Metadata
{
    private final ImportClientFactory importClientFactory;

    @Inject
    public ImportMetadata(ImportClientFactory importClientFactory)
    {
        this.importClientFactory = checkNotNull(importClientFactory, "importClientFactory is null");
    }

    @Override
    public FunctionInfo getFunction(QualifiedName name, List<TupleInfo.Type> parameterTypes)
    {
        throw new UnsupportedOperationException();
    }

    @Override
    public TableMetadata getTable(String catalogName, final String schemaName, final String tableName)
    {
        checkNotNull(catalogName, "catalogName is null");
        checkNotNull(schemaName, "schemaName is null");
        checkNotNull(tableName, "tableName is null");

        final ImportClient client = importClientFactory.getClient(catalogName);
        List<SchemaField> tableSchema = runWithRetryUnchecked(new Callable<List<SchemaField>>()
        {
            @Override
            public List<SchemaField> call()
                    throws Exception
            {
                return client.getTableSchema(schemaName, tableName);
            }
        });

        ImportTableHandle importTableHandle = new ImportTableHandle(catalogName, schemaName, tableName);

        List<ColumnMetadata> columns = convertToMetadata(catalogName, importTableHandle, tableSchema);

        return new TableMetadata(catalogName, schemaName, tableName, columns, importTableHandle);
    }

    private List<ColumnMetadata> convertToMetadata(final String sourceName, final ImportTableHandle importTableHandle, List<SchemaField> schemaFields)
    {
        return Lists.transform(schemaFields, new Function<SchemaField, ColumnMetadata>()
        {
            @Override
            public ColumnMetadata apply(SchemaField schemaField)
            {
                return new ColumnMetadata(
                        schemaField.getFieldName(),
                        ImportSchemaUtil.getTupleType(schemaField.getPrimitiveType()),
                        new ImportColumnHandle(sourceName, schemaField.getFieldName(), importTableHandle)
                );
            }
        });
    }

    @Override
    public void createTable(TableMetadata table)
    {
        throw new UnsupportedOperationException();
    }
}

@@ -2503,110 +2503,29 @@
     
-return new ColumnMetadata(%0A                        schemaField.getFieldName(),%0A                       
+TupleInfo.Type type =
  Imp
@@ -2582,17 +2582,17 @@
 eType())
-,
+;
 %0A       
@@ -2600,23 +2600,49 @@
         
-       
+ImportColumnHandle columnHandle =
  new Imp
@@ -2697,34 +2697,22 @@
 Name(), 
-importTableHandl
+typ
 e)
+;
 %0A       
@@ -2720,16 +2720,88 @@
         
+return new ColumnMetadata(schemaField.getFieldName(), type, columnHandle
 );%0A     

Fix previous commit