import io.airlift.json.JsonCodec; //导入依赖的package包/类
public static Map.Entry<SchemaTableName, RedisTableDescription> loadTpchTableDescription(
        JsonCodec<RedisTableDescription> tableDescriptionJsonCodec,
        SchemaTableName schemaTableName,
        String dataFormat)
        throws IOException
{
    RedisTableDescription tpchTemplate;
    try (InputStream data = RedisTestUtils.class.getResourceAsStream(format("/tpch/%s/%s.json", dataFormat, schemaTableName.getTableName()))) {
        tpchTemplate = tableDescriptionJsonCodec.fromJson(ByteStreams.toByteArray(data));
    }

    RedisTableDescription tableDescription = new RedisTableDescription(
            schemaTableName.getTableName(),
            schemaTableName.getSchemaName(),
            tpchTemplate.getKey(),
            tpchTemplate.getValue());

    return new AbstractMap.SimpleImmutableEntry<>(schemaTableName, tableDescription);
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public ClusterMemoryManager(
        @ForMemoryManager HttpClient httpClient,
        NodeManager nodeManager,
        LocationFactory locationFactory,
        MBeanExporter exporter,
        JsonCodec<MemoryInfo> memoryInfoCodec,
        JsonCodec<MemoryPoolAssignmentsRequest> assignmentsRequestJsonCodec,
        QueryIdGenerator queryIdGenerator,
        ServerConfig serverConfig,
        MemoryManagerConfig config)
{
    requireNonNull(config, "config is null");
    this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");
    this.locationFactory = requireNonNull(locationFactory, "locationFactory is null");
    this.httpClient = requireNonNull(httpClient, "httpClient is null");
    this.exporter = requireNonNull(exporter, "exporter is null");
    this.memoryInfoCodec = requireNonNull(memoryInfoCodec, "memoryInfoCodec is null");
    this.assignmentsRequestJsonCodec = requireNonNull(assignmentsRequestJsonCodec, "assignmentsRequestJsonCodec is null");
    this.maxQueryMemory = config.getMaxQueryMemory();
    this.coordinatorId = queryIdGenerator.getCoordinatorId();
    this.enabled = serverConfig.isCoordinator();
    this.killOnOutOfMemoryDelay = config.getKillOnOutOfMemoryDelay();
    this.killOnOutOfMemory = config.isKillOnOutOfMemory();
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public MetadataManager(FeaturesConfig featuresConfig,
        TypeManager typeManager,
        JsonCodec<ViewDefinition> viewCodec,
        BlockEncodingSerde blockEncodingSerde,
        SessionPropertyManager sessionPropertyManager,
        TablePropertyManager tablePropertyManager,
        TransactionManager transactionManager)
{
    functions = new FunctionRegistry(typeManager, blockEncodingSerde, featuresConfig.isExperimentalSyntaxEnabled());
    procedures = new ProcedureRegistry();
    this.typeManager = requireNonNull(typeManager, "types is null");
    this.viewCodec = requireNonNull(viewCodec, "viewCodec is null");
    this.blockEncodingSerde = requireNonNull(blockEncodingSerde, "blockEncodingSerde is null");
    this.sessionPropertyManager = requireNonNull(sessionPropertyManager, "sessionPropertyManager is null");
    this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
    this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");

    verifyComparableOrderableContract();
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
private static <T> JsonCodec<T> getJsonCodecForType(Type type)
{
    if (VarcharType.VARCHAR.equals(type)) {
        return (JsonCodec<T>) JSON_CODEC_FACTORY.jsonCodec(String.class);
    }
    if (BooleanType.BOOLEAN.equals(type)) {
        return (JsonCodec<T>) JSON_CODEC_FACTORY.jsonCodec(Boolean.class);
    }
    if (BigintType.BIGINT.equals(type)) {
        return (JsonCodec<T>) JSON_CODEC_FACTORY.jsonCodec(Long.class);
    }
    if (DoubleType.DOUBLE.equals(type)) {
        return (JsonCodec<T>) JSON_CODEC_FACTORY.jsonCodec(Double.class);
    }
    if (type instanceof ArrayType) {
        Type elementType = ((ArrayType) type).getElementType();
        return (JsonCodec<T>) JSON_CODEC_FACTORY.listJsonCodec(getJsonCodecForType(elementType));
    }
    if (type instanceof MapType) {
        Type keyType = ((MapType) type).getKeyType();
        Type valueType = ((MapType) type).getValueType();
        return (JsonCodec<T>) JSON_CODEC_FACTORY.mapJsonCodec(getMapKeyType(keyType), getJsonCodecForType(valueType));
    }
    throw new PrestoException(INVALID_SESSION_PROPERTY, format("Session property type %s is not supported", type));
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public ExecuteResource(
        HttpServerInfo serverInfo,
        AccessControl accessControl,
        SessionPropertyManager sessionPropertyManager,
        @ForExecute HttpClient httpClient,
        QueryIdGenerator queryIdGenerator,
        JsonCodec<QueryResults> queryResultsCodec)
{
    this.serverInfo = requireNonNull(serverInfo, "serverInfo is null");
    this.accessControl = requireNonNull(accessControl, "accessControl is null");
    this.sessionPropertyManager = requireNonNull(sessionPropertyManager, "sessionPropertyManager is null");
    this.httpClient = requireNonNull(httpClient, "httpClient is null");
    this.queryIdGenerator = requireNonNull(queryIdGenerator, "queryIdGenerator is null");
    this.queryResultsCodec = requireNonNull(queryResultsCodec, "queryResultsCodec is null");
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public HttpRemoteTaskFactory(QueryManagerConfig config,
        TaskManagerConfig taskConfig,
        @ForScheduler HttpClient httpClient,
        LocationFactory locationFactory,
        JsonCodec<TaskInfo> taskInfoCodec,
        JsonCodec<TaskUpdateRequest> taskUpdateRequestCodec)
{
    this.httpClient = httpClient;
    this.locationFactory = locationFactory;
    this.taskInfoCodec = taskInfoCodec;
    this.taskUpdateRequestCodec = taskUpdateRequestCodec;
    this.minErrorDuration = config.getRemoteTaskMinErrorDuration();
    this.taskInfoRefreshMaxWait = taskConfig.getInfoRefreshMaxWait();
    this.coreExecutor = newCachedThreadPool(daemonThreadsNamed("remote-task-callback-%s"));
    this.executor = new BoundedExecutor(coreExecutor, config.getRemoteTaskMaxCallbackThreads());
    this.executorMBean = new ThreadPoolExecutorMBean((ThreadPoolExecutor) coreExecutor);

    this.errorScheduledExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("remote-task-error-delay-%s"));
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Test
public void testSerialization()
        throws Exception
{
    String connectorId = "testid";
    SystemTableHandle tableHandle = new SystemTableHandle(connectorId, "xyz", "foo");
    SystemSplit expected = new SystemSplit(connectorId, tableHandle, HostAddress.fromParts("127.0.0.1", 0), TupleDomain.all());

    JsonCodec<SystemSplit> codec = jsonCodec(SystemSplit.class);
    SystemSplit actual = codec.fromJson(codec.toJson(expected));

    assertEquals(actual.getConnectorId(), expected.getConnectorId());
    assertEquals(actual.getTableHandle(), expected.getTableHandle());
    assertEquals(actual.getAddresses(), expected.getAddresses());
    assertEquals(actual.getConstraint(), expected.getConstraint());
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Test
public void testRoundTrip()
{
    ObjectMapperProvider objectMapperProvider = new ObjectMapperProvider();
    objectMapperProvider.setJsonDeserializers(ImmutableMap.<Class<?>, JsonDeserializer<?>>of(Type.class, new TypeDeserializer(new TypeRegistry())));
    JsonCodec<Signature> codec = new JsonCodecFactory(objectMapperProvider, true).jsonCodec(Signature.class);

    Signature expected = new Signature("function", SCALAR, StandardTypes.BIGINT, ImmutableList.of(StandardTypes.BOOLEAN, StandardTypes.DOUBLE, StandardTypes.VARCHAR));

    String json = codec.toJson(expected);
    Signature actual = codec.fromJson(json);

    assertEquals(actual.getName(), expected.getName());
    assertEquals(actual.getKind(), expected.getKind());
    assertEquals(actual.getReturnType(), expected.getReturnType());
    assertEquals(actual.getArgumentTypes(), expected.getArgumentTypes());
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
public CassandraSession(String connectorId,
        final Builder clusterBuilder,
        int fetchSizeForPartitionKeySelect,
        int limitForPartitionKeySelect,
        JsonCodec<List<ExtraColumnMetadata>> extraColumnMetadataCodec)
{
    this.connectorId = connectorId;
    this.fetchSizeForPartitionKeySelect = fetchSizeForPartitionKeySelect;
    this.limitForPartitionKeySelect = limitForPartitionKeySelect;
    this.extraColumnMetadataCodec = extraColumnMetadataCodec;

    sessionBySchema = CacheBuilder.newBuilder()
            .build(new CacheLoader<String, Session>()
            {
                @Override
                public Session load(String key)
                        throws Exception
                {
                    return clusterBuilder.build().connect();
                }
            });
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
public QueryRunner(
        ClientSession session,
        JsonCodec<QueryResults> queryResultsCodec,
        Optional<HostAndPort> socksProxy,
        Optional<String> keystorePath,
        Optional<String> keystorePassword,
        Optional<String> kerberosPrincipal,
        Optional<String> kerberosRemoteServiceName,
        boolean authenticationEnabled,
        KerberosConfig kerberosConfig)
{
    this.session = new AtomicReference<>(requireNonNull(session, "session is null"));
    this.queryResultsCodec = requireNonNull(queryResultsCodec, "queryResultsCodec is null");
    this.httpClient = new JettyHttpClient(
            getHttpClientConfig(socksProxy, keystorePath, keystorePassword, kerberosPrincipal, kerberosRemoteServiceName, authenticationEnabled),
            kerberosConfig,
            Optional.<JettyIoPool>empty(),
            ImmutableList.<HttpRequestFilter>of());
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
public StatementClient(HttpClient httpClient, JsonCodec<QueryResults> queryResultsCodec, ClientSession session, String query)
{
    requireNonNull(httpClient, "httpClient is null");
    requireNonNull(queryResultsCodec, "queryResultsCodec is null");
    requireNonNull(session, "session is null");
    requireNonNull(query, "query is null");

    this.httpClient = httpClient;
    this.responseHandler = createFullJsonResponseHandler(queryResultsCodec);
    this.debug = session.isDebug();
    this.timeZoneId = session.getTimeZoneId();
    this.query = query;
    this.requestTimeoutNanos = session.getClientRequestTimeout().roundTo(NANOSECONDS);

    Request request = buildQueryRequest(session, query);
    JsonResponse<QueryResults> response = httpClient.execute(request, responseHandler);

    if (response.getStatusCode() != HttpStatus.OK.code() || !response.hasValue()) {
        throw requestFailedException("starting query", request, response);
    }

    processResponse(response);
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public HivePageSinkProvider(
        HdfsEnvironment hdfsEnvironment,
        HiveMetastore metastore,
        PageIndexerFactory pageIndexerFactory,
        TypeManager typeManager,
        HiveClientConfig config,
        LocationService locationService,
        JsonCodec<PartitionUpdate> partitionUpdateCodec)
{
    this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
    this.metastore = requireNonNull(metastore, "metastore is null");
    this.pageIndexerFactory = requireNonNull(pageIndexerFactory, "pageIndexerFactory is null");
    this.typeManager = requireNonNull(typeManager, "typeManager is null");
    this.respectTableFormat = config.isRespectTableFormat();
    this.maxOpenPartitions = config.getMaxPartitionsPerWriter();
    this.immutablePartitions = config.isImmutablePartitions();
    this.locationService = requireNonNull(locationService, "locationService is null");
    this.partitionUpdateCodec = requireNonNull(partitionUpdateCodec, "partitionUpdateCodec is null");
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
DynamicAnnouncementFilter(
        JsonCodec<JsonAnnouncement> announcementCodec,
        @ForDynamicAnnouncements
                Map<String, Map<String, Supplier<String>>> propertiesMap)
{
    codec = announcementCodec;

    // Re-collect into immutable nested maps
    ImmutableMap.Builder<String, Map<String, Supplier<String>>> builder =
            ImmutableMap.builder();

    for (Entry<String, Map<String, Supplier<String>>> service
            : propertiesMap.entrySet()) {
        ImmutableMap<String, Supplier<String>> properties = service.getValue()
                .entrySet().stream()
                .collect(toImmutableMap(
                        Entry::getKey,
                        Entry::getValue,
                        (v1, v2) -> {
                            throw new IllegalArgumentException(
                                    "Service has two dynamic properties with the same name");
                        }));
        builder.put(service.getKey(), properties);
    }
    this.propertiesSuppliers = builder.build();
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
private void assertJsonRoundTrip(DataSize dataSize)
        throws IOException
{
    JsonCodec<DataSize> dataSizeCodec = JsonCodec.jsonCodec(DataSize.class);
    String json = dataSizeCodec.toJson(dataSize);
    DataSize dataSizeCopy = dataSizeCodec.fromJson(json);
    double delta = dataSize.toBytes() * 0.01;
    Assert.assertEquals(dataSize.toBytes(), dataSizeCopy.toBytes(), delta);
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
private void assertJsonRoundTrip(Duration duration)
        throws IOException
{
    JsonCodec<Duration> durationCodec = JsonCodec.jsonCodec(Duration.class);
    String json = durationCodec.toJson(duration);
    Duration durationCopy = durationCodec.fromJson(json);
    double delta = duration.getValue(MILLISECONDS) * 0.01;
    assertEquals(duration.getValue(MILLISECONDS), durationCopy.getValue(MILLISECONDS), delta);
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
KafkaTableDescriptionSupplier(KafkaConnectorConfig kafkaConnectorConfig,
        JsonCodec<KafkaTopicDescription> topicDescriptionCodec)
{
    this.topicDescriptionCodec = requireNonNull(topicDescriptionCodec, "topicDescriptionCodec is null");

    requireNonNull(kafkaConnectorConfig, "kafkaConfig is null");
    this.tableDescriptionDir = kafkaConnectorConfig.getTableDescriptionDir();
    this.defaultSchema = kafkaConnectorConfig.getDefaultSchema();
    this.tableNames = ImmutableSet.copyOf(kafkaConnectorConfig.getTableNames());
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
private static Map<SchemaTableName, KafkaTopicDescription> createTpchTopicDescriptions(Metadata metadata, Iterable<TpchTable<?>> tables)
        throws Exception
{
    JsonCodec<KafkaTopicDescription> topicDescriptionJsonCodec = new CodecSupplier<>(KafkaTopicDescription.class, metadata).get();

    ImmutableMap.Builder<SchemaTableName, KafkaTopicDescription> topicDescriptions = ImmutableMap.builder();
    for (TpchTable<?> table : tables) {
        String tableName = table.getTableName();
        SchemaTableName tpchTable = new SchemaTableName(TPCH_SCHEMA, tableName);

        topicDescriptions.put(loadTpchTopicDescription(topicDescriptionJsonCodec, tpchTable.toString(), tpchTable));
    }
    return topicDescriptions.build();
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
public static Map.Entry<SchemaTableName, KafkaTopicDescription> loadTpchTopicDescription(JsonCodec<KafkaTopicDescription> topicDescriptionJsonCodec, String topicName, SchemaTableName schemaTableName)
        throws IOException
{
    KafkaTopicDescription tpchTemplate = topicDescriptionJsonCodec.fromJson(ByteStreams.toByteArray(TestUtils.class.getResourceAsStream(format("/tpch/%s.json", schemaTableName.getTableName()))));

    return new AbstractMap.SimpleImmutableEntry<>(
            schemaTableName,
            new KafkaTopicDescription(schemaTableName.getTableName(), schemaTableName.getSchemaName(), topicName, tpchTemplate.getKey(), tpchTemplate.getMessage()));
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
@Inject
public ExampleClient(ExampleConfig config, JsonCodec<Map<String, List<ExampleTable>>> catalogCodec)
        throws IOException
{
    requireNonNull(config, "config is null");
    requireNonNull(catalogCodec, "catalogCodec is null");

    schemas = Suppliers.memoize(schemasSupplier(catalogCodec, config.getMetadata()));
}
 

import io.airlift.json.JsonCodec; //导入依赖的package包/类
private static Supplier<Map<String, Map<String, ExampleTable>>> schemasSupplier(final JsonCodec<Map<String, List<ExampleTable>>> catalogCodec, final URI metadataUri)
{
    return () -> {
        try {
            return lookupSchemas(metadataUri, catalogCodec);
        }
        catch (IOException e) {
            throw Throwables.propagate(e);
        }
    };
}