/*
 * 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.facebook.presto.connector.system;

import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorColumnHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorTableHandle;
import com.facebook.presto.spi.ConnectorTableMetadata;
import com.facebook.presto.spi.ReadOnlyConnectorMetadata;
import com.facebook.presto.spi.SchemaTableName;
import com.facebook.presto.spi.SchemaTablePrefix;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import static com.facebook.presto.connector.system.SystemColumnHandle.toSystemColumnHandles;
import static com.facebook.presto.metadata.MetadataUtil.findColumnMetadata;
import static com.facebook.presto.util.Types.checkType;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Predicates.compose;
import static com.google.common.base.Predicates.equalTo;
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.transform;

public class SystemTablesMetadata
        extends ReadOnlyConnectorMetadata
{
    private final ConcurrentMap<SchemaTableName, ConnectorTableMetadata> tables = new ConcurrentHashMap<>();

    public void addTable(ConnectorTableMetadata tableMetadata)
    {
        checkArgument(tables.putIfAbsent(tableMetadata.getTable(), tableMetadata) == null, "Table %s is already registered", tableMetadata.getTable());
    }

    private SystemTableHandle checkTableHandle(ConnectorTableHandle tableHandle)
    {
        SystemTableHandle systemTableHandle = checkType(tableHandle, SystemTableHandle.class, "tableHandle");
        checkArgument(tables.containsKey(systemTableHandle.getSchemaTableName()));
        return systemTableHandle;
    }

    @Override
    public List<String> listSchemaNames(ConnectorSession session)
    {
        // remove duplicates
        ImmutableSet<String> schemaNames = ImmutableSet.copyOf(transform(tables.keySet(), SchemaTableName::getSchemaName));
        return ImmutableList.copyOf(schemaNames);
    }

    @Override
    public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName)
    {
        if (!tables.containsKey(tableName)) {
            return null;
        }
        return new SystemTableHandle(tableName);
    }

    @Override
    public ConnectorTableMetadata getTableMetadata(ConnectorTableHandle tableHandle)
    {
        SystemTableHandle systemTableHandle = checkTableHandle(tableHandle);
        return tables.get(systemTableHandle.getSchemaTableName());
    }

    @Override
    public List<SchemaTableName> listTables(ConnectorSession session, String schemaNameOrNull)
    {
        if (schemaNameOrNull == null) {
            return ImmutableList.copyOf(tables.keySet());
        }

        return ImmutableList.copyOf(filter(tables.keySet(), compose(equalTo(schemaNameOrNull), SchemaTableName::getSchemaName)));
    }

    @Override
    public ConnectorColumnHandle getSampleWeightColumnHandle(ConnectorTableHandle tableHandle)
    {
        return null;
    }

    @Override
    public ColumnMetadata getColumnMetadata(ConnectorTableHandle tableHandle, ConnectorColumnHandle columnHandle)
    {
        SystemTableHandle systemTableHandle = checkTableHandle(tableHandle);
        ConnectorTableMetadata tableMetadata = tables.get(systemTableHandle.getSchemaTableName());

        String columnName = checkType(columnHandle, SystemColumnHandle.class, "columnHandle").getColumnName();

        ColumnMetadata columnMetadata = findColumnMetadata(tableMetadata, columnName);
        checkArgument(columnMetadata != null, "Column %s on table %s does not exist", columnName, tableMetadata.getTable());
        return columnMetadata;
    }

    @Override
    public Map<String, ConnectorColumnHandle> getColumnHandles(ConnectorTableHandle tableHandle)
    {
        SystemTableHandle systemTableHandle = checkTableHandle(tableHandle);

        return toSystemColumnHandles(tables.get(systemTableHandle.getSchemaTableName()));
    }

    @Override
    public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
    {
        checkNotNull(prefix, "prefix is null");
        ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> builder = ImmutableMap.builder();
        for (Entry<SchemaTableName, ConnectorTableMetadata> entry : tables.entrySet()) {
            if (prefix.matches(entry.getKey())) {
                builder.put(entry.getKey(), entry.getValue().getColumns());
            }
        }
        return builder.build();
    }
}

@@ -1114,55 +1114,8 @@
 Map;
-%0Aimport com.google.common.collect.ImmutableSet;
 %0A%0Aim
@@ -1441,24 +1441,100 @@
 mnMetadata;%0A
+import static com.facebook.presto.util.ImmutableCollectors.toImmutableList;%0A
 import stati
@@ -1711,241 +1711,8 @@
 ull;
-%0Aimport static com.google.common.base.Predicates.compose;%0Aimport static com.google.common.base.Predicates.equalTo;%0Aimport static com.google.common.collect.Iterables.filter;%0Aimport static com.google.common.collect.Iterables.transform;
 %0A%0Apu
@@ -2544,200 +2544,163 @@
     
-// remove duplicates%0A        ImmutableSet%3CString%3E schemaNames = ImmutableSet.copyOf(transform(tables.keySet(), SchemaTableName::getSchemaName));%0A        return ImmutableList.copyOf(schemaNames
+return tables.keySet().stream()%0A                .map(SchemaTableName::getSchemaName)%0A                .distinct()%0A                .collect(toImmutableList()
 );%0A 
@@ -3461,68 +3461,94 @@
 urn 
-Immu
 table
-List.copyOf(filter(tables.keySet(), compose(
+s.keySet().stream()%0A                .filter(table -%3E table.getSchemaName().
 equal
-To
+s
 (sch
@@ -3565,41 +3565,51 @@
 ull)
-, SchemaTableName::getSchemaName)
+)%0A                .collect(toImmutableList(
 ));%0A

Convert SystemTablesMetadata to streams API