import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testDefaults()
{
    assertRecordedDefaults(recordDefaults(ApacheThriftClientConfig.class)
            .setTransport(FRAMED)
            .setProtocol(BINARY)
            .setConnectTimeout(new Duration(500, MILLISECONDS))
            .setRequestTimeout(new Duration(1, MINUTES))
            .setSocksProxy(null)
            .setMaxFrameSize(new DataSize(16, MEGABYTE))
            .setMaxStringSize(new DataSize(16, MEGABYTE))
            .setSslEnabled(false)
            .setTrustCertificate(null)
            .setKey(null)
            .setKeyPassword(null));
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
public synchronized CompletableFuture<BufferResult> get(TaskId outputId, long startingSequenceId, DataSize maxSize)
{
    requireNonNull(outputId, "outputId is null");
    checkArgument(maxSize.toBytes() > 0, "maxSize must be at least 1 byte");

    // if no buffers can be added, and the requested buffer does not exist, return a closed empty result
    // this can happen with limit queries
    BufferState state = this.state.get();
    if (state != FAILED && !state.canAddBuffers() && namedBuffers.get(outputId) == null) {
        return completedFuture(emptyResults(taskInstanceId, 0, true));
    }

    // return a future for data
    GetBufferResult getBufferResult = new GetBufferResult(outputId, startingSequenceId, maxSize);
    stateChangeListeners.add(getBufferResult);
    updateState();
    return getBufferResult.getFuture();
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testExplicitPropertyMappings()
{
    Map<String, String> properties = new ImmutableMap.Builder<String, String>()
            .put("query.low-memory-killer.enabled", "true")
            .put("query.low-memory-killer.delay", "20s")
            .put("query.max-memory", "2GB")
            .put("query.max-memory-per-node", "2GB")
            .build();

    MemoryManagerConfig expected = new MemoryManagerConfig()
            .setKillOnOutOfMemory(true)
            .setKillOnOutOfMemoryDelay(new Duration(20, SECONDS))
            .setMaxQueryMemory(new DataSize(2, GIGABYTE))
            .setMaxQueryMemoryPerNode(new DataSize(2, GIGABYTE));

    assertFullMapping(properties, expected);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Override
public String toString()
{
    ToStringHelper stringHelper = toStringHelper(this)
            .add("tableId", "tableId")
            .add("shardId", shardId)
            .add("shardUuid", shardUuid)
            .add("rowCount", rowCount)
            .add("compressedSize", DataSize.succinctBytes(compressedSize))
            .add("uncompressedSize", DataSize.succinctBytes(uncompressedSize));

    if (rangeStart.isPresent()) {
        stringHelper.add("rangeStart", rangeStart.getAsLong());
    }
    if (rangeEnd.isPresent()) {
        stringHelper.add("rangeEnd", rangeEnd.getAsLong());
    }
    return stringHelper.toString();
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Override
public void run()
{
    try {
        stats.addQueuedTime(Duration.nanosSince(queuedTime));
        long start = System.nanoTime();

        backupStore.get().backupShard(uuid, source);
        stats.addCopyShardDataRate(new DataSize(source.length(), BYTE), Duration.nanosSince(start));
        stats.incrementBackupSuccess();
    }
    catch (Throwable t) {
        stats.incrementBackupFailure();
        throw Throwables.propagate(t);
    }
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testNonTemporalCompactionSetSimple()
        throws Exception
{
    CompactionSetCreator compactionSetCreator = new FileCompactionSetCreator(new DataSize(1, KILOBYTE), MAX_SHARD_ROWS);

    // compact into one shard
    Set<ShardMetadata> inputShards = ImmutableSet.of(
            shardWithSize(100),
            shardWithSize(100),
            shardWithSize(100));

    Set<CompactionSet> compactionSets = compactionSetCreator.createCompactionSets(1L, inputShards);
    assertEquals(compactionSets.size(), 1);
    assertEquals(getOnlyElement(compactionSets).getShardsToCompact(), inputShards);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testNonTemporalCompactionSet()
        throws Exception
{
    CompactionSetCreator compactionSetCreator = new FileCompactionSetCreator(new DataSize(100, BYTE), MAX_SHARD_ROWS);
    long tableId = 1L;

    // compact into two shards
    List<ShardMetadata> inputShards = ImmutableList.of(
            shardWithSize(70),
            shardWithSize(20),
            shardWithSize(30),
            shardWithSize(120));

    Set<CompactionSet> compactionSets = compactionSetCreator.createCompactionSets(tableId, ImmutableSet.copyOf(inputShards));
    assertEquals(compactionSets.size(), 2);
    Set<CompactionSet> expected = ImmutableSet.of(
            new CompactionSet(tableId, ImmutableSet.of(inputShards.get(0), inputShards.get(2))),
            new CompactionSet(tableId, ImmutableSet.of(inputShards.get(1))));
    assertEquals(compactionSets, expected);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testTemporalCompactionNoCompactionAcrossDays()
        throws Exception
{
    CompactionSetCreator compactionSetCreator = new TemporalCompactionSetCreator(new DataSize(100, BYTE), MAX_SHARD_ROWS, TIMESTAMP);
    long tableId = 1L;
    long day1 = Duration.ofDays(Duration.ofNanos(System.nanoTime()).toDays()).toMillis();
    long day2 = Duration.ofDays(Duration.ofMillis(day1).toDays() + 1).toMillis();

    // compact into two shards
    List<ShardMetadata> inputShards = ImmutableList.of(
            shardWithRange(10, day1, day1),
            shardWithRange(10, day2, day2),
            shardWithRange(10, day1, day1));

    Set<CompactionSet> actual = compactionSetCreator.createCompactionSets(tableId, ImmutableSet.copyOf(inputShards));
    assertEquals(actual.size(), 2);
    Set<CompactionSet> expected = ImmutableSet.of(
            new CompactionSet(tableId, ImmutableSet.of(inputShards.get(0), inputShards.get(2))),
            new CompactionSet(tableId, ImmutableSet.of(inputShards.get(1))));
    assertEquals(actual, expected);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testMaxFileSize()
        throws Exception
{
    List<Long> columnIds = ImmutableList.of(3L, 7L);
    List<Type> columnTypes = ImmutableList.<Type>of(BIGINT, VARCHAR);

    List<Page> pages = rowPagesBuilder(columnTypes)
            .row(123, "hello")
            .row(456, "bye")
            .build();

    // Set maxFileSize to 1 byte, so adding any page makes the StoragePageSink full
    OrcStorageManager manager = createOrcStorageManager(20, new DataSize(1, BYTE));
    StoragePageSink sink = createStoragePageSink(manager, columnIds, columnTypes);
    sink.appendPages(pages);
    assertTrue(sink.isFull());
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
public static OrcStorageManager createOrcStorageManager(
        StorageService storageService,
        Optional<BackupStore> backupStore,
        ShardRecoveryManager recoveryManager,
        ShardRecorder shardRecorder,
        int maxShardRows,
        DataSize maxFileSize)
{
    return new OrcStorageManager(
            CURRENT_NODE,
            storageService,
            backupStore,
            SHARD_DELTA_CODEC,
            READER_ATTRIBUTES,
            new BackupManager(backupStore, 1),
            recoveryManager,
            shardRecorder,
            new TypeRegistry(),
            CONNECTOR_ID,
            DELETION_THREADS,
            SHARD_RECOVERY_TIMEOUT,
            maxShardRows,
            maxFileSize);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
public static String formatDataRate(DataSize dataSize, Duration duration, boolean longForm)
{
    double rate = dataSize.toBytes() / duration.getValue(SECONDS);
    if (Double.isNaN(rate) || Double.isInfinite(rate)) {
        rate = 0;
    }

    String rateString = formatDataSize(new DataSize(rate, BYTE), false);
    if (longForm) {
        if (!rateString.endsWith("B")) {
            rateString += "B";
        }
        rateString += "/s";
    }
    return rateString;
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testMergeGap()
{
    List<DiskRange> consistent10ByteGap = ImmutableList.of(new DiskRange(100, 90), new DiskRange(200, 90), new DiskRange(300, 90));
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(0, BYTE), new DataSize(1, GIGABYTE)), consistent10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(9, BYTE), new DataSize(1, GIGABYTE)), consistent10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(1, GIGABYTE)), ImmutableList.of(new DiskRange(100, 290)));
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(100, BYTE), new DataSize(1, GIGABYTE)), ImmutableList.of(new DiskRange(100, 290)));

    List<DiskRange> middle10ByteGap = ImmutableList.of(new DiskRange(100, 80), new DiskRange(200, 90), new DiskRange(300, 80), new DiskRange(400, 90));
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(0, BYTE), new DataSize(1, GIGABYTE)), middle10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(9, BYTE), new DataSize(1, GIGABYTE)), middle10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(10, BYTE), new DataSize(1, GIGABYTE)),
            ImmutableList.of(new DiskRange(100, 80), new DiskRange(200, 180), new DiskRange(400, 90)));
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(100, BYTE), new DataSize(1, GIGABYTE)), ImmutableList.of(new DiskRange(100, 390)));
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testMergeMaxSize()
{
    List<DiskRange> consistent10ByteGap = ImmutableList.of(new DiskRange(100, 90), new DiskRange(200, 90), new DiskRange(300, 90));
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(0, BYTE)), consistent10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(100, BYTE)), consistent10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(190, BYTE)),
            ImmutableList.of(new DiskRange(100, 190), new DiskRange(300, 90)));
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(200, BYTE)),
            ImmutableList.of(new DiskRange(100, 190), new DiskRange(300, 90)));
    assertEquals(mergeAdjacentDiskRanges(consistent10ByteGap, new DataSize(10, BYTE), new DataSize(290, BYTE)), ImmutableList.of(new DiskRange(100, 290)));

    List<DiskRange> middle10ByteGap = ImmutableList.of(new DiskRange(100, 80), new DiskRange(200, 90), new DiskRange(300, 80), new DiskRange(400, 90));
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(0, BYTE), new DataSize(1, GIGABYTE)), middle10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(9, BYTE), new DataSize(1, GIGABYTE)), middle10ByteGap);
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(10, BYTE), new DataSize(1, GIGABYTE)),
            ImmutableList.of(new DiskRange(100, 80), new DiskRange(200, 180), new DiskRange(400, 90)));
    assertEquals(mergeAdjacentDiskRanges(middle10ByteGap, new DataSize(100, BYTE), new DataSize(1, GIGABYTE)), ImmutableList.of(new DiskRange(100, 390)));
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testIntegration()
        throws IOException
{
    // tiny file
    TestingOrcDataSource orcDataSource = new TestingOrcDataSource(
            new FileOrcDataSource(tempFile.getFile(), new DataSize(1, Unit.MEGABYTE), new DataSize(1, Unit.MEGABYTE), new DataSize(1, Unit.MEGABYTE)));
    doIntegration(orcDataSource, new DataSize(1, Unit.MEGABYTE), new DataSize(1, Unit.MEGABYTE));
    assertEquals(orcDataSource.getReadCount(), 1); // read entire file at once

    // tiny stripes
    orcDataSource = new TestingOrcDataSource(
            new FileOrcDataSource(tempFile.getFile(), new DataSize(1, Unit.MEGABYTE), new DataSize(1, Unit.MEGABYTE), new DataSize(1, Unit.MEGABYTE)));
    doIntegration(orcDataSource, new DataSize(400, Unit.KILOBYTE), new DataSize(400, Unit.KILOBYTE));
    assertEquals(orcDataSource.getReadCount(), 3); // footer, first few stripes, last few stripes
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
public IndexLookupSourceSupplier(
        Set<Integer> lookupSourceInputChannels,
        List<Integer> keyOutputChannels,
        Optional<Integer> keyOutputHashChannel,
        List<Type> outputTypes,
        IndexBuildDriverFactoryProvider indexBuildDriverFactoryProvider,
        DataSize maxIndexMemorySize,
        IndexJoinLookupStats stats,
        boolean shareIndexLoading)
{
    this.outputTypes = ImmutableList.copyOf(requireNonNull(outputTypes, "outputTypes is null"));

    if (shareIndexLoading) {
        IndexLoader shared = new IndexLoader(lookupSourceInputChannels, keyOutputChannels, keyOutputHashChannel, outputTypes, indexBuildDriverFactoryProvider, 10_000, maxIndexMemorySize, stats);
        this.indexLoaderSupplier = () -> shared;
    }
    else {
        this.indexLoaderSupplier = () -> new IndexLoader(lookupSourceInputChannels, keyOutputChannels, keyOutputHashChannel, outputTypes, indexBuildDriverFactoryProvider, 10_000, maxIndexMemorySize, stats);
    }
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
public ExchangeClient(
        BlockEncodingSerde blockEncodingSerde,
        DataSize maxBufferedBytes,
        DataSize maxResponseSize,
        int concurrentRequestMultiplier,
        Duration minErrorDuration,
        HttpClient httpClient,
        ScheduledExecutorService executor,
        SystemMemoryUsageListener systemMemoryUsageListener)
{
    this.blockEncodingSerde = blockEncodingSerde;
    this.maxBufferedBytes = maxBufferedBytes.toBytes();
    this.maxResponseSize = maxResponseSize;
    this.concurrentRequestMultiplier = concurrentRequestMultiplier;
    this.minErrorDuration = minErrorDuration;
    this.httpClient = httpClient;
    this.executor = executor;
    this.systemMemoryUsageListener = systemMemoryUsageListener;
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testExplicitPropertyMappings()
{
    Map<String, String> properties = new ImmutableMap.Builder<String, String>()
            .put("thrift.client.transport", "HEADER")
            .put("thrift.client.protocol", "COMPACT")
            .put("thrift.client.connect-timeout", "99ms")
            .put("thrift.client.request-timeout", "33m")
            .put("thrift.client.socks-proxy", "localhost:11")
            .put("thrift.client.max-frame-size", "55MB")
            .put("thrift.client.max-string-size", "66MB")
            .put("thrift.client.ssl.enabled", "true")
            .put("thrift.client.ssl.trust-certificate", "trust")
            .put("thrift.client.ssl.key", "key")
            .put("thrift.client.ssl.key-password", "key_password")
            .build();

    ApacheThriftClientConfig expected = new ApacheThriftClientConfig()
            .setTransport(HEADER)
            .setProtocol(COMPACT)
            .setConnectTimeout(new Duration(99, MILLISECONDS))
            .setRequestTimeout(new Duration(33, MINUTES))
            .setSocksProxy(HostAndPort.fromParts("localhost", 11))
            .setMaxFrameSize(new DataSize(55, MEGABYTE))
            .setMaxStringSize(new DataSize(66, MEGABYTE))
            .setSslEnabled(true)
            .setTrustCertificate(new File("trust"))
            .setKey(new File("key"))
            .setKeyPassword("key_password");

    assertFullMapping(properties, expected);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test
public void testExplicitPropertyMappings()
{
    Map<String, String> properties = new ImmutableMap.Builder<String, String>()
            .put("thrift.client.transport", "HEADER")
            .put("thrift.client.protocol", "COMPACT")
            .put("thrift.client.connect-timeout", "99ms")
            .put("thrift.client.request-timeout", "33m")
            .put("thrift.client.socks-proxy", "localhost:11")
            .put("thrift.client.max-frame-size", "55MB")
            .put("thrift.client.pool.enabled", "true")
            .put("thrift.client.ssl.enabled", "true")
            .put("thrift.client.ssl.trust-certificate", "trust")
            .put("thrift.client.ssl.key", "key")
            .put("thrift.client.ssl.key-password", "key_password")
            .put("thrift.client.ssl.session-cache-size", "678")
            .put("thrift.client.ssl.session-timeout", "78h")
            .put("thrift.client.ssl.ciphers", "some_cipher")
            .build();

    DriftNettyClientConfig expected = new DriftNettyClientConfig()
            .setTransport(HEADER)
            .setProtocol(COMPACT)
            .setConnectTimeout(new Duration(99, MILLISECONDS))
            .setRequestTimeout(new Duration(33, MINUTES))
            .setSocksProxy(HostAndPort.fromParts("localhost", 11))
            .setMaxFrameSize(new DataSize(55, MEGABYTE))
            .setPoolEnabled(true)
            .setSslEnabled(true)
            .setTrustCertificate(new File("trust"))
            .setKey(new File("key"))
            .setKeyPassword("key_password")
            .setSessionCacheSize(678)
            .setSessionTimeout(new Duration(78, HOURS))
            .setCiphers("some_cipher");

    assertFullMapping(properties, expected);
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Override
public Optional<DataSize> getValue()
{
    String prefix = getPrefix() + getName() + getSeparator();
    for (String arg : ManagementFactory.getRuntimeMXBean().getInputArguments()) {
        if (arg.startsWith(prefix)) {
            //return Optional.of(DataSize.succinctBytes(Long.valueOf(arg.substring(prefix.length())));
        }
    }
    return Optional.empty();
}
 

import io.airlift.units.DataSize; //导入依赖的package包/类
@Test(expectedExceptions = ExceededMemoryLimitException.class, expectedExceptionsMessageRegExp = "Query exceeded local memory limit of 10B")
public void testMemoryLimit()
        throws Exception
{
    List<Page> input = rowPagesBuilder(BIGINT, DOUBLE)
            .row(1, 0.1)
            .row(2, 0.2)
            .pageBreak()
            .row(-1, -0.1)
            .row(4, 0.4)
            .build();

    DriverContext driverContext = createTaskContext(executor, TEST_SESSION, new DataSize(10, Unit.BYTE))
            .addPipelineContext(true, true)
            .addDriverContext();

    WindowOperatorFactory operatorFactory = createFactoryUnbounded(
            ImmutableList.of(BIGINT, DOUBLE),
            Ints.asList(1),
            ROW_NUMBER,
            Ints.asList(),
            Ints.asList(0),
            ImmutableList.copyOf(new SortOrder[] {SortOrder.ASC_NULLS_LAST}));

    Operator operator = operatorFactory.createOperator(driverContext);

    toPages(operator, input);
}