@Override
    public void checkCanSelectFromColumns(ConnectorTransactionHandle transactionHandle, ConnectorIdentity identity, AccessControlContext context, SchemaTableName tableName, Set<Subfield> columnOrSubfieldNames)
    {
        Set<String> deniedColumns = new HashSet<>();
        for (String column : columnOrSubfieldNames.stream().map(column -> column.getRootName()).collect(toImmutableSet())) {
            if (!checkAccess(identity, tableName, column, HiveAccessType.SELECT)) {
                deniedColumns.add(column);
            }
        }
        if (deniedColumns.size() > 0) {
            denySelectColumns(tableName.getTableName(), columnOrSubfieldNames.stream().map(column -> column.getRootName()).collect(toImmutableSet()), format("Access denied - User [ %s ] does not have [SELECT] " +
                    "privilege on all mentioned columns of [ %s/%s ] ", identity.getUser(), tableName.getSchemaName(), tableName.getTableName()));
        }
    }
@Test
    public void testColumnLevelAccess()
    {
        ConnectorAccessControl accessControl = createRangerAccessControl("default-table-column-access.json", "user_groups.json");
        // 'analyst' group have read access {group - analyst, user - joe}
        accessControl.checkCanSelectFromColumns(TRANSACTION_HANDLE, user("joe"), CONTEXT, new SchemaTableName("foodmart", "salary"), ImmutableSet.of(new Subfield("salary_paid"), new Subfield("overtime_paid")));

        //  Access denied to others {group - readall, user - bob}
        assertDenied(() -> accessControl.checkCanSelectFromColumns(TRANSACTION_HANDLE, user("bob"), CONTEXT, new SchemaTableName("foodmart", "salary"), ImmutableSet.of(new Subfield("currency_id"), new Subfield("overtime_paid"))));
    }