diff --git a/hetu-docs/en/admin/properties.md b/hetu-docs/en/admin/properties.md index 10a379d73..8c57afc84 100644 --- a/hetu-docs/en/admin/properties.md +++ b/hetu-docs/en/admin/properties.md @@ -254,6 +254,15 @@ This section describes the most important config properties that may be used to > > Sets number of pages prefetched while reading from spilled files. + +### `experimental.spill-use-kryo-serialization` + +> - **Type:** `boolean` +> - **Default value:** `false` +> +> Enables Kryo based serialization for spill to disk, instead of default java serializer. + + ### `experimental.revocable-memory-selection-threshold` > - **Type:** `data size` diff --git a/hetu-hbase/pom.xml b/hetu-hbase/pom.xml index 7fa5d3130..e1f64269f 100644 --- a/hetu-hbase/pom.xml +++ b/hetu-hbase/pom.xml @@ -17,7 +17,7 @@ 13.0 2.0.1.Final 3.10 - 1.10.18 + 1.10.19 6.10 2.2.3 0.8.2 @@ -315,7 +315,7 @@ org.mockito - mockito-all + mockito-core ${version.mockito-all} test diff --git a/hetu-heuristic-index/pom.xml b/hetu-heuristic-index/pom.xml index c5d35d1f3..8936761ea 100644 --- a/hetu-heuristic-index/pom.xml +++ b/hetu-heuristic-index/pom.xml @@ -16,6 +16,7 @@ ${project.parent.basedir} false 1.7.30 + 2.0.2 @@ -100,36 +101,25 @@ org.mockito mockito-core - 3.5.13 + 1.10.19 test org.powermock powermock-core - 2.0.2 + ${powermock.version} test org.powermock powermock-module-testng - 2.0.2 + ${powermock.version} test org.powermock powermock-module-testng-common - 2.0.2 - test - - - org.powermock - powermock-api-mockito2 - 2.0.2 - test - - - org.objenesis - objenesis + ${powermock.version} test diff --git a/hetu-metastore/pom.xml b/hetu-metastore/pom.xml index a9a151d84..e47e3b985 100644 --- a/hetu-metastore/pom.xml +++ b/hetu-metastore/pom.xml @@ -149,8 +149,8 @@ org.mockito - mockito-all - 1.10.18 + mockito-core + 1.10.19 test diff --git a/hetu-transport/pom.xml b/hetu-transport/pom.xml index 7a4fc7fb0..4c525ac55 100644 --- a/hetu-transport/pom.xml +++ b/hetu-transport/pom.xml @@ -45,5 +45,10 @@ jackson-annotations provided + + com.esotericsoftware + kryo + 5.0.3 + \ No newline at end of file diff --git a/hetu-transport/src/main/java/io/hetu/core/transport/block/BlockSerdeUtil.java b/hetu-transport/src/main/java/io/hetu/core/transport/block/BlockSerdeUtil.java index 31495f637..745a1fa16 100644 --- a/hetu-transport/src/main/java/io/hetu/core/transport/block/BlockSerdeUtil.java +++ b/hetu-transport/src/main/java/io/hetu/core/transport/block/BlockSerdeUtil.java @@ -13,6 +13,10 @@ */ package io.hetu.core.transport.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.Slice; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; @@ -41,8 +45,18 @@ public final class BlockSerdeUtil return blockEncodingSerde.readBlock(input); } + public static Block readBlock(Kryo kryo, Serializer blockEncodingSerde, Input input) + { + return (Block) blockEncodingSerde.read(kryo, input, null); + } + public static void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput output, Block block) { blockEncodingSerde.writeBlock(output, block); } + + public static void writeBlock(Kryo kryo, Serializer blockEncodingSerde, Output output, Block block) + { + blockEncodingSerde.write(kryo, output, block); + } } diff --git a/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/KryoPageSerializer.java b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/KryoPageSerializer.java new file mode 100644 index 000000000..f3ca1d3de --- /dev/null +++ b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/KryoPageSerializer.java @@ -0,0 +1,113 @@ +/* + * Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved. + * 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 io.hetu.core.transport.execution.buffer; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import io.prestosql.spi.Page; +import io.prestosql.spi.block.Block; +import io.prestosql.spi.block.BlockEncodingSerde; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.util.Optional; +import java.util.Properties; + +import static com.google.common.base.Preconditions.checkArgument; +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Objects.requireNonNull; + +public class KryoPageSerializer + extends PagesSerde +{ + BlockEncodingSerde serde; + Serializer serializer = new Serializer() + { + @Override + public void write(Kryo kryo, Output output, Page page) + { + output.writeInt(page.getPositionCount()); + output.writeInt(page.getChannelCount()); + for (int channel = 0; channel < page.getChannelCount(); channel++) { + serde.writeBlock(output, page.getBlock(channel)); + } + + if (page.getPageMetadata().size() > 0) { + String pageProperties = page.getPageMetadata().toString(); + byte[] propertiesByte = pageProperties + .replaceAll(",", System.lineSeparator()) + .substring(1, pageProperties.length() - 1) + .getBytes(UTF_8); + output.writeInt(propertiesByte.length); + output.writeBytes(propertiesByte); + } + else { + output.writeInt(0); + } + } + + @Override + public Page read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + int numberOfBlocks = input.readInt(); + Block[] blocks = new Block[numberOfBlocks]; + for (int i = 0; i < blocks.length; i++) { + blocks[i] = serde.readBlock(input); + } + + int propSize = input.readInt(); + if (propSize > 0) { + byte[] pageMetadataBytes = input.readBytes(propSize); + Properties pros = new Properties(); + try { + pros.load(new ByteArrayInputStream(pageMetadataBytes)); + } + catch (IOException e) { + throw new UncheckedIOException(e); + } + + return new Page(positionCount, pros, blocks); + } + + return new Page(positionCount, blocks); + } + }; + + KryoPageSerializer(BlockEncodingSerde serde) + { + super(serde, Optional.empty(), Optional.empty(), Optional.empty()); + this.serde = requireNonNull(serde, "Serde Cannot be null"); + } + + @Override + public void serialize(OutputStream output, Page page) + { + checkArgument(output instanceof Output, "Page serializer does not support (" + output.getClass().getSimpleName() + ") for writing"); + serializer.write(null, (Output) output, page); + } + + @Override + public Page deserialize(InputStream input) + { + checkArgument(input instanceof Input, "Page serializer does not support (" + input.getClass().getSimpleName() + ") for reading"); + return serializer.read(null, (Input) input, Page.class); + } +} diff --git a/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeFactory.java b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeFactory.java index 6c06d8833..e19a02fa4 100644 --- a/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeFactory.java +++ b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeFactory.java @@ -35,17 +35,21 @@ public class PagesSerdeFactory public PagesSerde createPagesSerde() { - return createPagesSerdeInternal(Optional.empty(), false); + return createPagesSerdeInternal(Optional.empty(), false, false); } - public PagesSerde createPagesSerdeForSpill(Optional spillCipher, boolean useDirectSerde) + public PagesSerde createPagesSerdeForSpill(Optional spillCipher, boolean useDirectSerde, boolean useKryo) { - return createPagesSerdeInternal(spillCipher, useDirectSerde); + return createPagesSerdeInternal(spillCipher, useDirectSerde, useKryo); } - private PagesSerde createPagesSerdeInternal(Optional spillCipher, boolean useDirectSerde) + private PagesSerde createPagesSerdeInternal(Optional spillCipher, boolean useDirectSerde, boolean useKryo) { if (useDirectSerde) { + if (useKryo) { + return new KryoPageSerializer(blockEncodingSerde); + } + return new SliceStreamPageSerde(blockEncodingSerde, Optional.empty(), Optional.empty(), spillCipher); } diff --git a/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeUtil.java b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeUtil.java index 57cfc6443..71015138d 100644 --- a/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeUtil.java +++ b/hetu-transport/src/main/java/io/hetu/core/transport/execution/buffer/PagesSerdeUtil.java @@ -13,6 +13,10 @@ */ package io.hetu.core.transport.execution.buffer; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.collect.AbstractIterator; import io.airlift.slice.Slice; import io.airlift.slice.SliceInput; @@ -46,6 +50,25 @@ public class PagesSerdeUtil { } + static void writeRawPage(Kryo kryo, Page page, Output output, Serializer serde) + { + output.write(page.getChannelCount()); + for (int channel = 0; channel < page.getChannelCount(); channel++) { + writeBlock(kryo, serde, output, page.getBlock(channel)); + } + } + + static Page readRawPage(Kryo kryo, int positionCount, Input input, Serializer blockEncodingSerde) + { + int numberOfBlocks = input.readInt(); + Block[] blocks = new Block[numberOfBlocks]; + for (int i = 0; i < blocks.length; i++) { + blocks[i] = readBlock(kryo, blockEncodingSerde, input); + } + + return new Page(positionCount, blocks); + } + static void writeRawPage(Page page, SliceOutput output, BlockEncodingSerde serde) { output.writeInt(page.getChannelCount()); diff --git a/pom.xml b/pom.xml index 4d2c8e47b..7c62271a9 100644 --- a/pom.xml +++ b/pom.xml @@ -1491,7 +1491,7 @@ org.mockito mockito-core - 1.9.5 + 1.10.19 com.google.guava diff --git a/presto-main/pom.xml b/presto-main/pom.xml index 604789ce9..0fd1125b8 100644 --- a/presto-main/pom.xml +++ b/presto-main/pom.xml @@ -305,8 +305,21 @@ org.mockito - mockito-all + mockito-core + test 1.10.19 + + + org.objenesis + objenesis + + + + + + org.hamcrest + hamcrest-core + 1.1 test @@ -423,6 +436,12 @@ bcprov-jdk15on + + com.esotericsoftware + kryo + 5.0.3 + + org.testng @@ -466,6 +485,12 @@ presto-spi test-jar test + + + com.esotericsoftware + kryo + + diff --git a/presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java b/presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java index 57b689d99..bce51cdac 100644 --- a/presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java +++ b/presto-main/src/main/java/io/prestosql/metadata/FunctionAndTypeManager.java @@ -13,6 +13,7 @@ */ package io.prestosql.metadata; +import com.esotericsoftware.kryo.Kryo; import com.google.common.annotations.VisibleForTesting; import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; @@ -126,13 +127,15 @@ public class FunctionAndTypeManager private final LoadingCache functionCache; private final CacheStatsMBean cacheStatsMBean; private final ConcurrentMap blockEncodings = new ConcurrentHashMap<>(); + private final Kryo kryo; @Inject public FunctionAndTypeManager( TransactionManager transactionManager, FeaturesConfig featuresConfig, HandleResolver handleResolver, - Set types) + Set types, + Kryo kryo) { this.transactionManager = requireNonNull(transactionManager, "transactionManager is null"); this.builtInFunctionNamespaceManager = new BuiltInFunctionNamespaceManager(featuresConfig, this); @@ -148,6 +151,7 @@ public class FunctionAndTypeManager .build(CacheLoader.from(key -> resolveBuiltInFunction(key.functionName, fromTypeSignatures(key.parameterTypes)))); this.cacheStatsMBean = new CacheStatsMBean(functionCache); this.functionResolver = new FunctionResolver(this); + this.kryo = requireNonNull(kryo, "Kryo Object cannot be null"); // add the built-in BlockEncodings addBlockEncoding(new VariableWidthBlockEncoding()); @@ -168,7 +172,7 @@ public class FunctionAndTypeManager public static FunctionAndTypeManager createTestFunctionAndTypeManager() { - return new FunctionAndTypeManager(createTestTransactionManager(), new FeaturesConfig(), new HandleResolver(), ImmutableSet.of()); + return new FunctionAndTypeManager(createTestTransactionManager(), new FeaturesConfig(), new HandleResolver(), ImmutableSet.of(), new Kryo()); } public BlockEncoding getBlockEncoding(String encodingName) @@ -183,6 +187,11 @@ public class FunctionAndTypeManager return new InternalBlockEncodingSerde(this); } + public BlockEncodingSerde getBlockKryoEncodingSerde() + { + return new KryoBlockEncodingSerde(this, kryo); + } + public void addBlockEncoding(BlockEncoding blockEncoding) { requireNonNull(blockEncoding, "blockEncoding is null"); diff --git a/presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java b/presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java index e4ebcba3b..daf1ef743 100644 --- a/presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java +++ b/presto-main/src/main/java/io/prestosql/metadata/InternalBlockEncodingSerde.java @@ -75,7 +75,7 @@ final class InternalBlockEncodingSerde } } - private static String readLengthPrefixedString(SliceInput input) + protected static String readLengthPrefixedString(SliceInput input) { int length = input.readInt(); byte[] bytes = new byte[length]; diff --git a/presto-main/src/main/java/io/prestosql/metadata/KryoBlockEncodingSerde.java b/presto-main/src/main/java/io/prestosql/metadata/KryoBlockEncodingSerde.java new file mode 100644 index 000000000..e74c53122 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/metadata/KryoBlockEncodingSerde.java @@ -0,0 +1,137 @@ +/* + * Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved. + * 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 io.prestosql.metadata; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; +import io.prestosql.spi.block.AbstractBlockEncoding; +import io.prestosql.spi.block.Block; +import io.prestosql.spi.block.BlockEncoding; +import io.prestosql.spi.block.BlockEncodingSerde; + +import java.io.InputStream; +import java.io.OutputStream; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Objects.requireNonNull; + +public class KryoBlockEncodingSerde + implements BlockEncodingSerde +{ + private final FunctionAndTypeManager functionAndTypeManager; + private final Kryo kryo; + + public KryoBlockEncodingSerde(FunctionAndTypeManager metadata, Kryo kryo) + { + this.functionAndTypeManager = requireNonNull(metadata, "metadata is null"); + this.kryo = kryo; /*Todo(nitin) make it singleton inject time initializer */ + } + + public Kryo getKryo() + { + return kryo; + } + + @Override + public Object getContext() + { + return getKryo(); + } + + /** + * Read a block encoding from the input. + * + * @param inputStream + */ + @Override + public Block readBlock(InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, + "This interface should not be called in this flow"); + } + + Input input = (Input) inputStream; + String encodingName; + // read the encoding name + encodingName = readLengthPrefixedString((Input) input); + + // look up the encoding factory + BlockEncoding blockEncoding = functionAndTypeManager.getBlockEncoding(encodingName); + Serializer serializer = getSerializerFromBlockEncoding(blockEncoding); + if (serializer == null) { + return (Block) blockEncoding.readBlock(this, inputStream); + } + return (Block) serializer.read(kryo, input, null); + } + + /** + * Write a blockEncoding to the output. + * + * @param outputStream + * @param block + */ + @Override + public void writeBlock(OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, + "This interface should not be called in this flow"); + } + + Output output = (Output) outputStream; + + String encodingName = block.getEncodingName(); + BlockEncoding blockEncoding = functionAndTypeManager.getBlockEncoding(encodingName); + Serializer> serializer = getSerializerFromBlockEncoding(blockEncoding); + + // write the name to the output + writeLengthPrefixedString(output, encodingName); + + if (serializer == null) { + blockEncoding.writeBlock(this, outputStream, block); + } + else { + // write the block to the output + serializer.write(kryo, output, block); + } + } + + private Serializer> getSerializerFromBlockEncoding(BlockEncoding blockEncoding) + { + if (blockEncoding instanceof AbstractBlockEncoding) { + return (Serializer>) blockEncoding; + } + return null; + } + + private static String readLengthPrefixedString(Input input) + { + int length = input.readInt(); + byte[] bytes = input.readBytes(length); + return new String(bytes, UTF_8); + } + + private static void writeLengthPrefixedString(Output output, String value) + { + byte[] bytes = value.getBytes(UTF_8); + output.writeInt(bytes.length); + output.writeBytes(bytes); + } +} diff --git a/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java b/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java index 000cb0cf8..f1e271044 100755 --- a/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java +++ b/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java @@ -13,6 +13,7 @@ */ package io.prestosql.metadata; +import com.esotericsoftware.kryo.Kryo; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Joiner; import com.google.common.collect.HashMultimap; @@ -201,7 +202,7 @@ public final class MetadataManager public static MetadataManager createTestMetadataManager(TransactionManager transactionManager, FeaturesConfig featuresConfig) { return new MetadataManager( - new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of()), + new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of(), new Kryo()), featuresConfig, new SessionPropertyManager(), new SchemaPropertyManager(), diff --git a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java index 42a0ea5e9..cdbdb8120 100644 --- a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java +++ b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java @@ -13,6 +13,7 @@ */ package io.prestosql.server; +import com.esotericsoftware.kryo.Kryo; import com.google.common.collect.ImmutableList; import com.google.inject.Binder; import com.google.inject.Key; @@ -439,6 +440,8 @@ public class ServerMainModule jsonBinder(binder).addDeserializerBinding(Type.class).to(TypeDeserializer.class); newSetBinder(binder, Type.class); + binder.bind(Kryo.class).in(Scopes.SINGLETON); + // split manager binder.bind(SplitManager.class).in(Scopes.SINGLETON); diff --git a/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpiller.java b/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpiller.java index ffb3654b9..6e29d1fae 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpiller.java +++ b/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpiller.java @@ -13,6 +13,8 @@ */ package io.prestosql.spiller; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Stopwatch; import com.google.common.collect.AbstractIterator; @@ -89,6 +91,7 @@ public class FileSingleStreamSpiller private long spilledPagesInMemorySize; private ListenableFuture spillInProgress = Futures.immediateFuture(null); private boolean useDirectSerde; + private boolean useKryo; private boolean compressionEnabled; private Optional spillCipher; private byte[] cipherIV; @@ -107,7 +110,8 @@ public class FileSingleStreamSpiller Optional spillCipher, boolean compressionEnabled, boolean useDirectSerde, - int spillPrefetchReadPages) + int spillPrefetchReadPages, + boolean useKryo) { this.serde = requireNonNull(serde, "serde is null"); this.executor = requireNonNull(executor, "executor is null"); @@ -136,6 +140,7 @@ public class FileSingleStreamSpiller } this.spillPrefetchReadPages = spillPrefetchReadPages; this.useDirectSerde = useDirectSerde; + this.useKryo = useKryo; this.compressionEnabled = compressionEnabled; this.spillCipher = spillCipher; } @@ -206,9 +211,14 @@ public class FileSingleStreamSpiller System.arraycopy(tmpCipherIV, 0, this.cipherIV, 0, tmpCipherIV.length); tmpOutputStream = new CipherOutputStream(tmpOutputStream, cipher); } + if (compressionEnabled) { tmpOutputStream = new SnappyFramedOutputStream(tmpOutputStream); } + + if (useKryo) { + return new Output(tmpOutputStream, bufferSize); + } return new OutputStreamSliceOutput(tmpOutputStream, bufferSize); } @@ -240,14 +250,23 @@ public class FileSingleStreamSpiller if (spillCipher.isPresent()) { tmpInputStream = new CipherInputStream(tmpInputStream, spillCipher.get().getDecryptionCipher(cipherIV)); } + if (compressionEnabled) { tmpInputStream = new SnappyFramedInputStream(tmpInputStream); } + + if (useKryo) { + return new Input(tmpInputStream, bufferSize); + } return new InputStreamSliceInput(tmpInputStream, bufferSize); } private Predicate getStreamEndOfData() { + if (useKryo) { + return (input) -> ((Input) input).end(); + } + return (input) -> !((InputStreamSliceInput) input).isReadable(); } @@ -348,6 +367,7 @@ public class FileSingleStreamSpiller state.targetFile = this.targetFile.getFilePath().toAbsolutePath().toString(); state.compressionEnabled = this.compressionEnabled; state.useDirectSerde = this.useDirectSerde; + state.useKryo = this.useKryo; return state; } @@ -371,6 +391,7 @@ public class FileSingleStreamSpiller } this.compressionEnabled = myState.compressionEnabled; this.useDirectSerde = myState.useDirectSerde; + this.useKryo = myState.useKryo; } catch (IOException e) { throw new UncheckedIOException(e); @@ -386,5 +407,6 @@ public class FileSingleStreamSpiller private String targetFile; private boolean compressionEnabled; private boolean useDirectSerde; + private boolean useKryo; } } diff --git a/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpillerFactory.java b/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpillerFactory.java index 8628b83c7..9a56b901a 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpillerFactory.java +++ b/presto-main/src/main/java/io/prestosql/spiller/FileSingleStreamSpillerFactory.java @@ -21,6 +21,7 @@ import io.airlift.log.Logger; import io.hetu.core.transport.execution.buffer.PagesSerde; import io.hetu.core.transport.execution.buffer.PagesSerdeFactory; import io.prestosql.memory.context.LocalMemoryContext; +import io.prestosql.metadata.KryoBlockEncodingSerde; import io.prestosql.metadata.Metadata; import io.prestosql.operator.SpillContext; import io.prestosql.spi.PrestoException; @@ -39,6 +40,7 @@ import java.nio.file.Path; import java.util.List; import java.util.Optional; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator; import static io.airlift.concurrent.Threads.daemonThreadsNamed; import static io.prestosql.spi.StandardErrorCode.OUT_OF_SPILL_SPACE; @@ -68,6 +70,7 @@ public class FileSingleStreamSpillerFactory private final double maxUsedSpaceThreshold; private final boolean spillEncryptionEnabled; private final boolean spillDirectSerdeEnabled; + private final boolean useKryo; private final boolean spillCompressionEnabled; private int roundRobinIndex; private int spillPrefetchReadPages; @@ -79,14 +82,15 @@ public class FileSingleStreamSpillerFactory listeningDecorator(newFixedThreadPool( requireNonNull(featuresConfig, "featuresConfig is null").getSpillerThreads(), daemonThreadsNamed("binary-spiller-%s"))), - requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockEncodingSerde(), + requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillUseKryoSerialization() ? requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockKryoEncodingSerde() : requireNonNull(metadata, "metadata is null").getFunctionAndTypeManager().getBlockEncodingSerde(), spillerStats, requireNonNull(featuresConfig, "featuresConfig is null").getSpillerSpillPaths(), requireNonNull(featuresConfig, "featuresConfig is null").getSpillMaxUsedSpaceThreshold(), requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillCompressionEnabled(), requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillEncryptionEnabled(), requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillDirectSerdeEnabled(), - requireNonNull(nodeSpillConfig, "featuresConfig is null").getSpillPrefetchReadPages()); + requireNonNull(nodeSpillConfig, "featuresConfig is null").getSpillPrefetchReadPages(), + requireNonNull(nodeSpillConfig, "nodeSpillConfig is null").isSpillUseKryoSerialization()); } @VisibleForTesting @@ -101,6 +105,28 @@ public class FileSingleStreamSpillerFactory boolean spillDirectSerdeEnabled, int spillPrefetchReadPages) { + this(executor, blockEncodingSerde, spillerStats, spillPaths, maxUsedSpaceThreshold, + spillCompressionEnabled, spillEncryptionEnabled, spillDirectSerdeEnabled, + spillPrefetchReadPages, false); + } + + @VisibleForTesting + public FileSingleStreamSpillerFactory( + ListeningExecutorService executor, + BlockEncodingSerde blockEncodingSerde, + SpillerStats spillerStats, + List spillPaths, + double maxUsedSpaceThreshold, + boolean spillCompressionEnabled, + boolean spillEncryptionEnabled, + boolean spillDirectSerdeEnabled, + int spillPrefetchReadPages, + boolean useKryo) + { + checkArgument(!(blockEncodingSerde instanceof KryoBlockEncodingSerde) + || (blockEncodingSerde instanceof KryoBlockEncodingSerde && spillDirectSerdeEnabled), + "Kryo serialization should enable DirectSpill"); + this.serdeFactory = new PagesSerdeFactory(blockEncodingSerde, spillCompressionEnabled); this.executor = requireNonNull(executor, "executor is null"); this.spillerStats = requireNonNull(spillerStats, "spillerStats can not be null"); @@ -125,6 +151,7 @@ public class FileSingleStreamSpillerFactory this.roundRobinIndex = 0; this.spillDirectSerdeEnabled = spillDirectSerdeEnabled; this.spillPrefetchReadPages = spillPrefetchReadPages; + this.useKryo = useKryo; } @PostConstruct @@ -164,8 +191,8 @@ public class FileSingleStreamSpillerFactory if (spillEncryptionEnabled) { spillCipher = Optional.of(new AesSpillCipher()); } - PagesSerde serde = serdeFactory.createPagesSerdeForSpill(spillCipher, spillDirectSerdeEnabled); - return new FileSingleStreamSpiller(serde, executor, getNextSpillPath(), spillerStats, spillContext, memoryContext, spillCipher, spillCompressionEnabled, spillDirectSerdeEnabled, spillPrefetchReadPages); + PagesSerde serde = serdeFactory.createPagesSerdeForSpill(spillCipher, spillDirectSerdeEnabled, useKryo); + return new FileSingleStreamSpiller(serde, executor, getNextSpillPath(), spillerStats, spillContext, memoryContext, spillCipher, spillCompressionEnabled, spillDirectSerdeEnabled, spillPrefetchReadPages, useKryo); } private synchronized Path getNextSpillPath() diff --git a/presto-main/src/main/java/io/prestosql/spiller/NodeSpillConfig.java b/presto-main/src/main/java/io/prestosql/spiller/NodeSpillConfig.java index 1fc01072e..5e4f5c5f7 100644 --- a/presto-main/src/main/java/io/prestosql/spiller/NodeSpillConfig.java +++ b/presto-main/src/main/java/io/prestosql/spiller/NodeSpillConfig.java @@ -31,6 +31,7 @@ public class NodeSpillConfig private boolean spillDirectSerdeEnabled; private int spillPrefetchReadPages = 1; + private boolean spillUseKryoSerialization; @NotNull public DataSize getMaxSpillPerNode() @@ -107,4 +108,16 @@ public class NodeSpillConfig this.spillPrefetchReadPages = spillPrefetchedReadPages; return this; } + + public boolean isSpillUseKryoSerialization() + { + return spillUseKryoSerialization; + } + + @Config("experimental.spill-use-kryo-serialization") + public NodeSpillConfig setSpillUseKryoSerialization(boolean spillUseKryoSerialization) + { + this.spillUseKryoSerialization = spillUseKryoSerialization; + return this; + } } diff --git a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java index bcb28f8eb..4b050e135 100644 --- a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java +++ b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java @@ -13,6 +13,7 @@ */ package io.prestosql.testing; +import com.esotericsoftware.kryo.Kryo; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; @@ -340,7 +341,7 @@ public class LocalQueryRunner this.planOptimizerManager = new ConnectorPlanOptimizerManager(); this.metadata = new MetadataManager( - new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of()), + new FunctionAndTypeManager(transactionManager, featuresConfig, new HandleResolver(), ImmutableSet.of(), new Kryo()), featuresConfig, // new HetuConfig object passed, if split filtering is needed in the runner, a modified HetuConfig object with filter settings manually set must be used. new SessionPropertyManager(new SystemSessionProperties(new QueryManagerConfig(), taskManagerConfig, new MemoryManagerConfig(), featuresConfig, new HetuConfig(), new SnapshotConfig())), diff --git a/presto-main/src/test/java/io/prestosql/operator/TestGroupByHash.java b/presto-main/src/test/java/io/prestosql/operator/TestGroupByHash.java index cb8738684..ab129e22d 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestGroupByHash.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestGroupByHash.java @@ -331,7 +331,7 @@ public class TestGroupByHash expectedMapping.put("hashCollisions", 37L); expectedMapping.put("expectedHashCollisions", 0.0); expectedMapping.put("preallocatedMemoryInBytes", 0L); - expectedMapping.put("currentPageSizeInBytes", 4732L); + expectedMapping.put("currentPageSizeInBytes", 4740L); return expectedMapping; } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestHashAggregationOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestHashAggregationOperator.java index 6e3909159..aaeb32751 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestHashAggregationOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestHashAggregationOperator.java @@ -317,7 +317,7 @@ public class TestHashAggregationOperator //TODO-cp-I2DSGQ: change expectedMapping after implementation of operatorContext capture expectedMapping.put("operatorContext", 0); expectedMapping.put("aggregationBuilder", aggregationBuilderMapping); - expectedMapping.put("memoryContext", 10355419L); + expectedMapping.put("memoryContext", 10675419L); expectedMapping.put("inputProcessed", true); expectedMapping.put("finishing", false); expectedMapping.put("finished", false); @@ -405,7 +405,7 @@ public class TestHashAggregationOperator aggregation3Array0.put("capacity", 40960); aggregation3Array0.put("segments", 40); aggregation3Array1.put("array", blockObjectBigArrayState); - aggregation3Array1.put("sizeOfBlocks", 2504108L); + aggregation3Array1.put("sizeOfBlocks", 2824108L); blockObjectBigArrayState.put("array", 1); blockObjectBigArrayState.put("capacity", 40960); aggregation3StateList.add(39999L); diff --git a/presto-main/src/test/java/io/prestosql/operator/TestOrderByOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestOrderByOperator.java index 5dffb9c4f..98029f431 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestOrderByOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestOrderByOperator.java @@ -266,7 +266,7 @@ public class TestOrderByOperator Map expectedMapping = new HashMap<>(); expectedMapping.put("operatorContext", 0); expectedMapping.put("revocableMemoryContext", 0L); - expectedMapping.put("localUserMemoryContext", 8828L); + expectedMapping.put("localUserMemoryContext", 8844L); return expectedMapping; } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestPagesIndex.java b/presto-main/src/test/java/io/prestosql/operator/TestPagesIndex.java index c57fcd73c..2469acc98 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestPagesIndex.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestPagesIndex.java @@ -80,8 +80,8 @@ public class TestPagesIndex expectedMapping.put("valueAddresses", valueAddresses); expectedMapping.put("nextBlockToCompact", 0); expectedMapping.put("positionCount", 7); - expectedMapping.put("pagesMemorySize", 3852L); - expectedMapping.put("estimatedSize", 12396L); + expectedMapping.put("pagesMemorySize", 3860L); + expectedMapping.put("estimatedSize", 12404L); return expectedMapping; } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestStreamingAggregationOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestStreamingAggregationOperator.java index 42c9fad73..772a51983 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestStreamingAggregationOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestStreamingAggregationOperator.java @@ -178,7 +178,7 @@ public class TestStreamingAggregationOperator Map expectedMapping = new HashMap<>(); expectedMapping.put("operatorContext", 0); expectedMapping.put("systemMemoryContext", 0L); - expectedMapping.put("userMemoryContext", 2236L); + expectedMapping.put("userMemoryContext", 2244L); expectedMapping.put("finishing", false); return expectedMapping; } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestTopNOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestTopNOperator.java index 1430d5ac3..6dfcb5af3 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestTopNOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestTopNOperator.java @@ -179,7 +179,7 @@ public class TestTopNOperator expectedMapping.put("operatorContext", 0); expectedMapping.put("workProcessorOperator", workProcessorOperatorMapping); - workProcessorOperatorMapping.put("localUserMemoryContext", 19152L); + workProcessorOperatorMapping.put("localUserMemoryContext", 19168L); workProcessorOperatorMapping.put("topNBuilder", topNBuilderMapping); workProcessorOperatorMapping.put("outputIterator", false); @@ -187,7 +187,7 @@ public class TestTopNOperator topNBuilderMapping.put("groupedRows", groupedRowsMapping); topNBuilderMapping.put("pageReferences", pageReferencesMapping); topNBuilderMapping.put("emptyPageReferenceSlots", emptyPageReferenceSlots); - topNBuilderMapping.put("memorySizeInBytes", 2496L); + topNBuilderMapping.put("memorySizeInBytes", 2512L); topNBuilderMapping.put("currentPageCount", 2); groupedRowsMapping.put("array", Object[][].class); diff --git a/presto-main/src/test/java/io/prestosql/spiller/TestFileSingleStreamSpiller.java b/presto-main/src/test/java/io/prestosql/spiller/TestFileSingleStreamSpiller.java index 451aa5bff..af35cf3a5 100644 --- a/presto-main/src/test/java/io/prestosql/spiller/TestFileSingleStreamSpiller.java +++ b/presto-main/src/test/java/io/prestosql/spiller/TestFileSingleStreamSpiller.java @@ -186,29 +186,55 @@ public class TestFileSingleStreamSpiller public void testSpillWithSingleFile() throws Exception { - assertSpillBenchmark(false, false, "1GB", 1, false); + assertSpillBenchmark(false, false, "1GB", 1, false, false); } @Test public void testSpillWithMultiFile() throws Exception { - assertSpillBenchmark(false, false, "2MB", 500, false); + assertSpillBenchmark(false, false, "2MB", 500, false, false); } - private void assertSpillBenchmark(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde) + @Test + public void testSpillWithSingleFileWithKryo() + throws Exception + { + assertSpillBenchmark(false, false, "1GB", 1, true, true); + } + + @Test + public void testSpillWithMultiFileWithKryo() + throws Exception + { + assertSpillBenchmark(false, false, "2MB", 2, true, true); + } + + @Test + public void testSpillWithSingleSpillerConsolidatedWithoutWorkProcessor() + throws Exception + { + assertSpillBenchmark(false, false, "1GB", 1, false, false); + assertSpillBenchmark(false, false, "1GB", 1, true, true); + assertSpillBenchmark(false, false, "2MB", 512, false, false); + assertSpillBenchmark(false, false, "2MB", 512, true, true); + } + + private void assertSpillBenchmark(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useKryo, boolean useDirectSerde) throws Exception { List spillers = new ArrayList<>(); FileSingleStreamSpillerFactory spillerFactory = new FileSingleStreamSpillerFactory( executorBenchmark, // executor won't be closed, because we don't call destroy() on the spiller factory - createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(), + (useKryo) ? createTestMetadataManager().getFunctionAndTypeManager().getBlockKryoEncodingSerde() : createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(), new SpillerStats(), ImmutableList.of(spillPath.toPath()), 1.0, compression, encryption, - useDirectSerde, 1); + useDirectSerde, + 1, + useKryo); LocalMemoryContext memoryContext = newSimpleAggregatedMemoryContext().newLocalMemoryContext("test"); long startTime = System.currentTimeMillis(); Stopwatch spillTimer = Stopwatch.createStarted(); @@ -285,81 +311,81 @@ public class TestFileSingleStreamSpiller return new Page(col1.build(), col2.build()); } + @Test + public void testSpillWithSingleSpillerConsolidated() + throws Exception + { + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 1, false); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1, true); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 1, false); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 1, true); + } + @Test public void testSpillWithSingleSpillerConsolidatedWithCompression() throws Exception { - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 25); + assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, false, 1, false); + assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1, false); } @Test public void testSpillWithSingleSpillerConsolidatedWithCompressionMultiFile() throws Exception { - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 25); + assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, false, 1, false); + assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "2MB", 512, true, 1, false); } @Test public void testSpillWithSingleSpillerConsolidatedWithoutCompression() throws Exception { - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 25); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, false, 25, false); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 25, false); } @Test public void testSpillWithSingleSpillerConsolidatedWithoutCompressionMultiFile() throws Exception { - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 25); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, false, 25, false); + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "2MB", 512, true, 25, false); } @Test public void testSpillWithSingleSpillerConsolidatedWithEncryption() throws Exception { - assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 1); - assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 1); - - assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 25); - - assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 25); - assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 25); + assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, false, 25, false); + assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 25, false); } - private void assertSpillBenchmarkReadingUsingWorkProcessor(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde, int spillPrefetchReadPages) + @Test + public void testSpillWithSingleSpillerConsolidatedDirectWriteCompareWithWorkProcessorWithKryo() + throws Exception + { + assertSpillBenchmarkReadingUsingWorkProcessor(false, false, "1GB", 1, true, 1, true); + assertSpillBenchmarkReadingUsingWorkProcessor(true, false, "1GB", 1, true, 1, true); + assertSpillBenchmarkReadingUsingWorkProcessor(false, true, "1GB", 1, true, 1, true); + assertSpillBenchmarkReadingUsingWorkProcessor(true, true, "1GB", 1, true, 1, true); + } + + private void assertSpillBenchmarkReadingUsingWorkProcessor(boolean compression, boolean encryption, String pageSize, int fileCount, boolean useDirectSerde, int spillPrefetchReadPages, boolean useKryo) throws Exception { List spillers = new ArrayList<>(); FileSingleStreamSpillerFactory spillerFactory = new FileSingleStreamSpillerFactory( executorBenchmark, // executor won't be closed, because we don't call destroy() on the spiller factory - createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(), + (useKryo) ? createTestMetadataManager().getFunctionAndTypeManager().getBlockKryoEncodingSerde() : createTestMetadataManager().getFunctionAndTypeManager().getBlockEncodingSerde(), new SpillerStats(), ImmutableList.of(spillPath.toPath()), 1.0, compression, encryption, useDirectSerde, - spillPrefetchReadPages); + spillPrefetchReadPages, + useKryo); LocalMemoryContext memoryContext = newSimpleAggregatedMemoryContext().newLocalMemoryContext("test"); long startTime = System.currentTimeMillis(); @@ -420,13 +446,9 @@ public class TestFileSingleStreamSpiller } spillTimer.stop(); long timeToUnspill = spillTimer.elapsed(TimeUnit.MILLISECONDS); - /* - * System.out.println(String.format("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d] --> Spill: %8d, Unspill: %8d", - * encryption, compression, useDirectSerde, pageSize, fileCount, timeToSpill, timeToUnspill, humanReadableByteCountBin(spilledDiskSize))); - */ log.debug("TimeTakenReadingFromSpill = " + (System.currentTimeMillis() - startTime)); - log.info("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d] --> Spill: %8d, Unspill: %8d, DiskUsed: %5s", - encryption, compression, useDirectSerde, pageSize, fileCount, + log.info("[isEncrypted: %6s, isCompressed: %6s, isDirect: %6s, SpillSize: %4s, SpillFiles: %4d, useKryo: %6s] --> Spill: %8d, Unspill: %8d, DiskUsed: %5s", + encryption, compression, useDirectSerde, pageSize, fileCount, useKryo, timeToSpill, timeToUnspill, humanReadableByteCountBin(spilledDiskSize)); spillers.stream().forEach(spiller -> spiller.close()); assertEquals(listFiles(spillPath.toPath()).size(), 0); diff --git a/presto-main/src/test/java/io/prestosql/spiller/TestNodeSpillConfig.java b/presto-main/src/test/java/io/prestosql/spiller/TestNodeSpillConfig.java index 8261c5d8e..2829ad98e 100644 --- a/presto-main/src/test/java/io/prestosql/spiller/TestNodeSpillConfig.java +++ b/presto-main/src/test/java/io/prestosql/spiller/TestNodeSpillConfig.java @@ -36,7 +36,8 @@ public class TestNodeSpillConfig .setSpillCompressionEnabled(false) .setSpillEncryptionEnabled(false) .setSpillDirectSerdeEnabled(false) - .setSpillPrefetchReadPages(1)); + .setSpillPrefetchReadPages(1) + .setSpillUseKryoSerialization(false)); } @Test @@ -49,6 +50,7 @@ public class TestNodeSpillConfig .put("experimental.spill-encryption-enabled", "true") .put("experimental.spill-direct-serde-enabled", "true") .put("experimental.spill-prefetch-read-pages", "25") + .put("experimental.spill-use-kryo-serialization", "true") .build(); NodeSpillConfig expected = new NodeSpillConfig() @@ -57,7 +59,8 @@ public class TestNodeSpillConfig .setSpillCompressionEnabled(true) .setSpillEncryptionEnabled(true) .setSpillDirectSerdeEnabled(true) - .setSpillPrefetchReadPages(25); + .setSpillPrefetchReadPages(25) + .setSpillUseKryoSerialization(true); assertFullMapping(properties, expected); } diff --git a/presto-main/src/test/java/io/prestosql/spiller/TestSpillCipherPagesSerde.java b/presto-main/src/test/java/io/prestosql/spiller/TestSpillCipherPagesSerde.java index b4ab3c0d8..427b4152e 100644 --- a/presto-main/src/test/java/io/prestosql/spiller/TestSpillCipherPagesSerde.java +++ b/presto-main/src/test/java/io/prestosql/spiller/TestSpillCipherPagesSerde.java @@ -41,7 +41,7 @@ public class TestSpillCipherPagesSerde public void test() { SpillCipher cipher = new AesSpillCipher(); - PagesSerde serde = TESTING_SERDE_FACTORY.createPagesSerdeForSpill(Optional.of(cipher), false); + PagesSerde serde = TESTING_SERDE_FACTORY.createPagesSerdeForSpill(Optional.of(cipher), false, false); List types = ImmutableList.of(VARCHAR); Page emptyPage = new Page(VARCHAR.createBlockBuilder(null, 0).build()); assertPageEquals(types, serde.deserialize(serde.serialize(emptyPage)), emptyPage); diff --git a/presto-spi/pom.xml b/presto-spi/pom.xml index 0da7d9ff3..ddeef2a26 100644 --- a/presto-spi/pom.xml +++ b/presto-spi/pom.xml @@ -145,5 +145,10 @@ io.airlift joni + + com.esotericsoftware + kryo + 5.0.3 + diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/AbstractBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/AbstractBlockEncoding.java new file mode 100644 index 000000000..14fb5e73e --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/block/AbstractBlockEncoding.java @@ -0,0 +1,23 @@ +/* + * Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved. + * 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 io.prestosql.spi.block; + +import com.esotericsoftware.kryo.Serializer; + +public abstract class AbstractBlockEncoding + extends Serializer + implements BlockEncoding +{ +} diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/ArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/ArrayBlockEncoding.java index a2630431a..e3d224ffc 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/ArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/ArrayBlockEncoding.java @@ -13,9 +13,16 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; import io.airlift.slice.Slices; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.ArrayBlock.createArrayBlockInternal; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; @@ -65,4 +72,55 @@ public class ArrayBlockEncoding boolean[] valueIsNull = decodeNullBits(sliceInput, positionCount).orElseGet(() -> new boolean[positionCount]); return createArrayBlockInternal(0, positionCount, valueIsNull, offsets, values); } + + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleMap ReadBlock"); + } + + Input input = (Input) inputStream; + Block values = blockEncodingSerde.readBlock(input); + int positionCount = input.readInt(); + int[] offsets = input.readInts(positionCount); + + boolean[] valueIsNull = null; + if (input.readBoolean()) { + valueIsNull = input.readBooleans(positionCount); + } + + return createArrayBlockInternal(0, positionCount, valueIsNull, offsets, values); + } + + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock"); + } + + Output output = (Output) outputStream; + AbstractArrayBlock arrayBlock = (AbstractArrayBlock) block; + + int positionCount = arrayBlock.getPositionCount(); + + int offsetBase = arrayBlock.getOffsetBase(); + int[] offsets = arrayBlock.getOffsets(); + + int valuesStartOffset = offsets[offsetBase]; + int valuesEndOffset = offsets[offsetBase + positionCount]; + Block values = arrayBlock.getRawElementBlock().getRegion(valuesStartOffset, valuesEndOffset - valuesStartOffset); + blockEncodingSerde.writeBlock(output, values); + + output.writeInt(positionCount); + for (int position = 0; position < positionCount + 1; position++) { + output.writeInt(offsets[offsetBase + position] - valuesStartOffset); + } + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(arrayBlock.getValueIsNull(), offsetBase, positionCount); + } + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncoding.java index b873308ef..8759111ac 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncoding.java @@ -16,6 +16,8 @@ package io.prestosql.spi.block; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import java.io.InputStream; +import java.io.OutputStream; import java.util.Optional; public interface BlockEncoding @@ -36,6 +38,17 @@ public interface BlockEncoding */ void writeBlock(BlockEncodingSerde blockEncodingSerde, SliceOutput sliceOutput, Block block); + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + */ + Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input); + + /** + * Write the specified block to the specified output + */ + void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block); + /** * This method allows the implementor to specify a replacement object that will be serialized instead of the original one. */ diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncodingSerde.java b/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncodingSerde.java index c3b1cd858..6f9e9b3d1 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncodingSerde.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/BlockEncodingSerde.java @@ -15,16 +15,49 @@ package io.prestosql.spi.block; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; public interface BlockEncodingSerde { /** * Read a block encoding from the input. */ - Block readBlock(SliceInput input); + default Block readBlock(InputStream input) + { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported"); + } /** * Write a blockEncoding to the output. */ - void writeBlock(SliceOutput output, Block block); + default void writeBlock(OutputStream output, Block block) + { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported"); + } + + /** + * Read a block encoding from the input. + */ + default Block readBlock(SliceInput input) + { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported"); + } + + /** + * Write a blockEncoding to the output. + */ + default void writeBlock(SliceOutput output, Block block) + { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Not supported"); + } + + /* give out context object if any */ + default Object getContext() + { + return null; + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java index 370e919bf..db69176a0 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java @@ -35,11 +35,11 @@ public class ByteArrayBlock { private static final int INSTANCE_SIZE = ClassLayout.parseClass(ByteArrayBlock.class).instanceSize(); - private final int arrayOffset; + protected final int arrayOffset; private final int positionCount; @Nullable - private final boolean[] valueIsNull; - private final byte[] values; + protected final boolean[] valueIsNull; + protected final byte[] values; private final long sizeInBytes; private final long retainedSizeInBytes; diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlockEncoding.java index 56bd608d9..930f57853 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlockEncoding.java @@ -13,14 +13,22 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class ByteArrayBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "BYTE_ARRAY"; @@ -61,4 +69,63 @@ public class ByteArrayBlockEncoding return new ByteArrayBlock(0, positionCount, valueIsNull, values); } + + @Override + public void write(Kryo kryo, Output output, ByteArrayBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + output.writeBytes(block.values, block.arrayOffset, positionCount); + } + + @Override + public ByteArrayBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + boolean[] valueIsNull = null; + if (input.readBoolean()) { + valueIsNull = input.readBooleans(positionCount); + } + byte[] values = input.readBytes(positionCount); + return new ByteArrayBlock(0, positionCount, valueIsNull, values); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for ByteArrayBlock"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, ByteArrayBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for ByteArrayBlock"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (ByteArrayBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java index 7748cd9ad..9c6cabdc9 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java @@ -41,8 +41,8 @@ public class DictionaryBlock private final int positionCount; private final Block dictionary; - private final int idsOffset; - private final int[] ids; + protected final int idsOffset; + protected final int[] ids; private final long retainedSizeInBytes; private volatile long sizeInBytes = -1; private volatile long logicalSizeInBytes = -1; diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlockEncoding.java index 70f278313..8ff1f02ae 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlockEncoding.java @@ -13,9 +13,16 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; import io.airlift.slice.Slices; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; public class DictionaryBlockEncoding implements BlockEncoding @@ -76,4 +83,57 @@ public class DictionaryBlockEncoding // TODO: fix DictionaryBlock so that dictionaryIsCompacted can be set to true when the underlying block over-retains memory. return new DictionaryBlock(positionCount, dictionaryBlock, ids, false, new DictionaryId(mostSignificantBits, leastSignificantBits, sequenceId)); } + + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for Dictionary ReadBlock"); + } + + Input input = (Input) inputStream; + int positionCount = input.readInt(); + + // dictionary + Block dictionaryBlock = blockEncodingSerde.readBlock(input); + + // ids + int[] ids = input.readInts(positionCount); + + // instance id + long mostSignificantBits = input.readLong(); + long leastSignificantBits = input.readLong(); + long sequenceId = input.readLong(); + + return new DictionaryBlock(positionCount, dictionaryBlock, ids, false, new DictionaryId(mostSignificantBits, leastSignificantBits, sequenceId)); + } + + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock"); + } + + Output output = (Output) outputStream; + DictionaryBlock dictionaryBlock = (DictionaryBlock) block; + + dictionaryBlock = dictionaryBlock.compact(); + + // positionCount + int positionCount = dictionaryBlock.getPositionCount(); + output.writeInt(positionCount); + + // dictionary + Block dictionary = dictionaryBlock.getDictionary(); + blockEncodingSerde.writeBlock(output, dictionary); + + // ids + output.writeInts(dictionaryBlock.ids, dictionaryBlock.idsOffset, positionCount); + + // instance id + output.writeLong(dictionaryBlock.getDictionarySourceId().getMostSignificantBits()); + output.writeLong(dictionaryBlock.getDictionarySourceId().getLeastSignificantBits()); + output.writeLong(dictionaryBlock.getDictionarySourceId().getSequenceId()); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java index 64eb57c84..6f31599b8 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java @@ -38,11 +38,11 @@ public class Int128ArrayBlock private static final int INSTANCE_SIZE = ClassLayout.parseClass(Int128ArrayBlock.class).instanceSize(); public static final int INT128_BYTES = Long.BYTES + Long.BYTES; - private final int positionOffset; + protected final int positionOffset; private final int positionCount; @Nullable - private final boolean[] valueIsNull; - private final long[] values; + protected final boolean[] valueIsNull; + protected final long[] values; private final long sizeInBytes; private final long retainedSizeInBytes; diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlockEncoding.java index b224e2ea6..114ae8a87 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlockEncoding.java @@ -13,14 +13,22 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class Int128ArrayBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "INT128_ARRAY"; @@ -63,4 +71,64 @@ public class Int128ArrayBlockEncoding return new Int128ArrayBlock(0, positionCount, valueIsNull, values); } + + @Override + public void write(Kryo kryo, Output output, Int128ArrayBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + output.writeBoolean(block.mayHaveNull()); + + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + output.writeLongs(block.values, block.positionOffset, positionCount * 2); + } + + @Override + public Int128ArrayBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + boolean[] valuesIsNull = null; + if (input.readBoolean()) { + valuesIsNull = input.readBooleans(positionCount); + } + long[] values = input.readLongs(positionCount * 2); + return new Int128ArrayBlock(0, positionCount, valuesIsNull, values); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for Int128"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, Int128ArrayBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for Int128"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (Int128ArrayBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java index ecec2b593..2868fd050 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java @@ -35,11 +35,11 @@ public class IntArrayBlock { private static final int INSTANCE_SIZE = ClassLayout.parseClass(IntArrayBlock.class).instanceSize(); - private final int arrayOffset; + protected final int arrayOffset; private final int positionCount; @Nullable - private final boolean[] valueIsNull; - private final int[] values; + protected final boolean[] valueIsNull; + protected final int[] values; private final long sizeInBytes; private final long retainedSizeInBytes; @@ -136,12 +136,6 @@ public class IntArrayBlock return getInt(position, offset); } - @Override - public String getString(int position, int offset, int length) - { - return String.valueOf(getInt(position, offset)); - } - @Override public boolean mayHaveNull() { diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlockEncoding.java index c63e99065..4602a96d6 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlockEncoding.java @@ -13,14 +13,22 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class IntArrayBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "INT_ARRAY"; @@ -61,4 +69,64 @@ public class IntArrayBlockEncoding return new IntArrayBlock(0, positionCount, valueIsNull, values); } + + @Override + public void write(Kryo kryo, Output output, IntArrayBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + output.writeInts(block.values, block.arrayOffset, positionCount); + } + + @Override + public IntArrayBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + boolean[] valuesIsNull = null; + if (input.readBoolean()) { + valuesIsNull = input.readBooleans(positionCount); + } + int[] values = input.readInts(positionCount); + return new IntArrayBlock(0, positionCount, valuesIsNull, values); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for IntArrayBlock"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, IntArrayBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for IntArrayBlock"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (IntArrayBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/LazyBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/LazyBlockEncoding.java index c9d00f711..6938c97f2 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/LazyBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/LazyBlockEncoding.java @@ -16,6 +16,8 @@ package io.prestosql.spi.block; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import java.io.InputStream; +import java.io.OutputStream; import java.util.Optional; public class LazyBlockEncoding @@ -45,6 +47,18 @@ public class LazyBlockEncoding throw new UnsupportedOperationException(); } + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + throw new UnsupportedOperationException(); + } + + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + throw new UnsupportedOperationException(); + } + @Override public Optional replacementBlockForWrite(Block block) { diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java index 60be2c314..2b47a8942 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java @@ -39,11 +39,11 @@ public class LongArrayBlock //can we intro operations at block level?, for example join of blocks? private static final int INSTANCE_SIZE = ClassLayout.parseClass(LongArrayBlock.class).instanceSize(); - private final int arrayOffset; + protected final int arrayOffset; private final int positionCount; @Nullable - private final boolean[] valueIsNull; - private final long[] values; //change to use offheap --> accessible by RDMA + protected final boolean[] valueIsNull; + protected final long[] values; //change to use offheap --> accessible by RDMA private final long sizeInBytes; private final long retainedSizeInBytes; diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlockEncoding.java index 548a8142e..619e48cf0 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlockEncoding.java @@ -13,14 +13,22 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class LongArrayBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "LONG_ARRAY"; @@ -61,4 +69,64 @@ public class LongArrayBlockEncoding return new LongArrayBlock(0, positionCount, valueIsNull, values); } + + @Override + public void write(Kryo kryo, Output output, LongArrayBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + output.writeLongs(block.values, block.arrayOffset, positionCount); + } + + @Override + public LongArrayBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + boolean[] valuesIsNull = null; + if (input.readBoolean()) { + valuesIsNull = input.readBooleans(positionCount); + } + long[] values = input.readLongs(positionCount); + return new LongArrayBlock(0, positionCount, valuesIsNull, values); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for LongArrayBlock"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, LongArrayBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for LongArrayBlock"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (LongArrayBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/MapBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/MapBlockEncoding.java index b1ba666b3..f94ebfd5a 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/MapBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/MapBlockEncoding.java @@ -14,12 +14,18 @@ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; import io.prestosql.spi.type.MapType; import io.prestosql.spi.type.TypeManager; import io.prestosql.spi.type.TypeSerde; +import java.io.InputStream; +import java.io.OutputStream; import java.util.Optional; import static io.airlift.slice.Slices.wrappedIntArray; @@ -96,4 +102,73 @@ public class MapBlockEncoding Optional mapIsNull = EncoderUtil.decodeNullBits(sliceInput, positionCount); return createMapBlockInternal(mapType, 0, positionCount, mapIsNull, offsets, keyBlock, valueBlock, hashTable); } + + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for MapBlock ReadBlock"); + } + + Input input = (Input) inputStream; + MapType mapType = (MapType) TypeSerde.readType(typeManager, input); + + Block keyBlock = blockEncodingSerde.readBlock(input); + Block valueBlock = blockEncodingSerde.readBlock(input); + + int hashTableSize = input.readInt(); + int[] hashTable = input.readInts(hashTableSize); + + if (keyBlock.getPositionCount() != valueBlock.getPositionCount() || keyBlock.getPositionCount() * HASH_MULTIPLIER != hashTable.length) { + throw new IllegalArgumentException( + format("Deserialized MapBlock violates invariants: key %d, value %d, hash %d", keyBlock.getPositionCount(), valueBlock.getPositionCount(), hashTable.length)); + } + + int positionCount = input.readInt(); + int[] offsets = input.readInts(positionCount + 1); + Optional mapIsNull = Optional.empty(); + if (input.readBoolean()) { + mapIsNull = Optional.of(input.readBooleans(positionCount)); + } + + return createMapBlockInternal(mapType, 0, positionCount, mapIsNull, offsets, keyBlock, valueBlock, hashTable); + } + + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock"); + } + + Output output = (Output) outputStream; + AbstractMapBlock mapBlock = (AbstractMapBlock) block; + + int positionCount = mapBlock.getPositionCount(); + + int offsetBase = mapBlock.getOffsetBase(); + int[] offsets = mapBlock.getOffsets(); + int[] hashTable = mapBlock.getHashTables(); + + int entriesStartOffset = offsets[offsetBase]; + int entriesEndOffset = offsets[offsetBase + positionCount]; + + TypeSerde.writeType(output, mapBlock.mapType); + + blockEncodingSerde.writeBlock(output, mapBlock.getRawKeyBlock().getRegion(entriesStartOffset, entriesEndOffset - entriesStartOffset)); + blockEncodingSerde.writeBlock(output, mapBlock.getRawValueBlock().getRegion(entriesStartOffset, entriesEndOffset - entriesStartOffset)); + + output.writeInt((entriesEndOffset - entriesStartOffset) * HASH_MULTIPLIER); + output.writeInts(hashTable, entriesStartOffset * HASH_MULTIPLIER, (entriesEndOffset - entriesStartOffset) * HASH_MULTIPLIER); + + output.writeInt(positionCount); + for (int position = 0; position < positionCount + 1; position++) { + output.writeInt(offsets[offsetBase + position] - entriesStartOffset); + } + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(mapBlock.getMapIsNull(), mapBlock.getOffsetBase(), positionCount); + } + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/RowBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/RowBlockEncoding.java index ecb5e9b33..5e856e336 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/RowBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/RowBlockEncoding.java @@ -14,8 +14,15 @@ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.airlift.slice.Slices.wrappedIntArray; import static io.prestosql.spi.block.RowBlock.createRowBlockInternal; @@ -71,4 +78,66 @@ public class RowBlockEncoding boolean[] rowIsNull = EncoderUtil.decodeNullBits(sliceInput, positionCount).orElseGet(() -> new boolean[positionCount]); return createRowBlockInternal(0, positionCount, rowIsNull, fieldBlockOffsets, fieldBlocks); } + + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleRow ReadBlock"); + } + + Input input = (Input) inputStream; + int numFields = input.readInt(); + Block[] fieldBlocks = new Block[numFields]; + for (int i = 0; i < fieldBlocks.length; i++) { + fieldBlocks[i] = blockEncodingSerde.readBlock(input); + } + + int positionCount = input.readInt(); + int[] fieldBlockOffsets = input.readInts(positionCount); + boolean[] rowIsNull; + + if (input.readBoolean()) { + rowIsNull = input.readBooleans(positionCount); + } + else { + rowIsNull = new boolean[positionCount]; + } + + return createRowBlockInternal(0, positionCount, rowIsNull, fieldBlockOffsets, fieldBlocks); + } + + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for RowBlock WriteBlock"); + } + + Output output = (Output) outputStream; + AbstractRowBlock rowBlock = (AbstractRowBlock) block; + int numFields = rowBlock.numFields; + + int positionCount = rowBlock.getPositionCount(); + + int offsetBase = rowBlock.getOffsetBase(); + int[] fieldBlockOffsets = rowBlock.getFieldBlockOffsets(); + int startFieldBlockOffset = fieldBlockOffsets[offsetBase]; + int endFieldBlockOffset = fieldBlockOffsets[offsetBase + positionCount]; + + output.writeInt(numFields); + for (int i = 0; i < numFields; i++) { + blockEncodingSerde.writeBlock(output, rowBlock.getRawFieldBlocks()[i].getRegion(startFieldBlockOffset, endFieldBlockOffset - startFieldBlockOffset)); + } + + output.writeInt(positionCount); + for (int position = 0; position < positionCount + 1; position++) { + output.writeInt(fieldBlockOffsets[offsetBase + position] - startFieldBlockOffset); + } + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(rowBlock.getRowIsNull(), offsetBase, positionCount); + } + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/RunLengthBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/RunLengthBlockEncoding.java index 28f709522..12d15d42d 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/RunLengthBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/RunLengthBlockEncoding.java @@ -13,8 +13,15 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; public class RunLengthBlockEncoding implements BlockEncoding @@ -50,4 +57,49 @@ public class RunLengthBlockEncoding return new RunLengthEncodedBlock(value, positionCount); } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param inputStream + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for RLE ReadBlock"); + } + + Input input = (Input) inputStream; + int positionCount = input.readInt(); + + Block value = blockEncodingSerde.readBlock(input); + return new RunLengthEncodedBlock(value, positionCount); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param outputStream + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for RLE WriteBlock"); + } + + RunLengthEncodedBlock rleBlock = (RunLengthEncodedBlock) block; + Output output = (Output) outputStream; + + // write the run length + output.writeInt(rleBlock.getPositionCount()); + + // write the value + blockEncodingSerde.writeBlock(output, rleBlock.getValue()); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java index 9da05b214..7472e5e83 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlock.java @@ -35,11 +35,11 @@ public class ShortArrayBlock { private static final int INSTANCE_SIZE = ClassLayout.parseClass(ShortArrayBlock.class).instanceSize(); - private final int arrayOffset; + protected final int arrayOffset; private final int positionCount; @Nullable - private final boolean[] valueIsNull; - private final short[] values; + protected final boolean[] valueIsNull; + protected final short[] values; private final long sizeInBytes; private final long retainedSizeInBytes; diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlockEncoding.java index bf5dd116c..2f046839b 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/ShortArrayBlockEncoding.java @@ -13,14 +13,22 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class ShortArrayBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "SHORT_ARRAY"; @@ -61,4 +69,64 @@ public class ShortArrayBlockEncoding return new ShortArrayBlock(0, positionCount, valueIsNull, values); } + + @Override + public void write(Kryo kryo, Output output, ShortArrayBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + output.writeShorts(block.values, block.arrayOffset, positionCount); + } + + @Override + public ShortArrayBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + boolean[] valuesIsNull = null; + if (input.readBoolean()) { + valuesIsNull = input.readBooleans(positionCount); + } + short[] values = input.readShorts(positionCount); + return new ShortArrayBlock(0, positionCount, valuesIsNull, values); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for ShortArrayBlock"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, ShortArrayBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for ShortArrayBlock"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (ShortArrayBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/SingleMapBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/SingleMapBlockEncoding.java index f568c3ef5..ad71253b7 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/SingleMapBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/SingleMapBlockEncoding.java @@ -14,12 +14,19 @@ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; import io.prestosql.spi.type.MapType; import io.prestosql.spi.type.TypeManager; import io.prestosql.spi.type.TypeSerde; +import java.io.InputStream; +import java.io.OutputStream; + import static io.airlift.slice.Slices.wrappedIntArray; import static io.prestosql.spi.block.AbstractMapBlock.HASH_MULTIPLIER; import static java.lang.String.format; @@ -76,4 +83,65 @@ public class SingleMapBlockEncoding return new SingleMapBlock(mapType, 0, keyBlock.getPositionCount() * 2, keyBlock, valueBlock, hashTable); } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param inputStream + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleMap ReadBlock"); + } + + Input input = (Input) inputStream; + MapType mapType = (MapType) TypeSerde.readType(typeManager, input); + + Block keyBlock = blockEncodingSerde.readBlock(input); + Block valueBlock = blockEncodingSerde.readBlock(input); + + int hashTableSize = input.readInt(); + int[] hashTable = input.readInts(hashTableSize); + + if (keyBlock.getPositionCount() != valueBlock.getPositionCount() + || keyBlock.getPositionCount() * HASH_MULTIPLIER != hashTable.length) { + throw new IllegalArgumentException( + format("Deserialized SingleMapBlock violates invariants: key %d, value %d, hash %d", keyBlock.getPositionCount(), valueBlock.getPositionCount(), hashTable.length)); + } + + return new SingleMapBlock(mapType, 0, keyBlock.getPositionCount() * 2, keyBlock, valueBlock, hashTable); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param outputStream + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleMap WriteBlock"); + } + + Output output = (Output) outputStream; + SingleMapBlock singleMapBlock = (SingleMapBlock) block; + TypeSerde.writeType(output, singleMapBlock.mapType); + + int offset = singleMapBlock.getOffset(); + int positionCount = singleMapBlock.getPositionCount(); + + blockEncodingSerde.writeBlock(output, singleMapBlock.getRawKeyBlock().getRegion(offset / 2, positionCount / 2)); + blockEncodingSerde.writeBlock(output, singleMapBlock.getRawValueBlock().getRegion(offset / 2, positionCount / 2)); + + int[] hashTable = singleMapBlock.getHashTable(); + output.writeInt(positionCount / 2 * HASH_MULTIPLIER); + output.writeInts(hashTable, offset / 2 * HASH_MULTIPLIER, positionCount / 2 * HASH_MULTIPLIER); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockEncoding.java index 706881585..82078c51c 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockEncoding.java @@ -14,8 +14,15 @@ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; public class SingleRowBlockEncoding implements BlockEncoding @@ -50,4 +57,53 @@ public class SingleRowBlockEncoding } return new SingleRowBlock(0, fieldBlocks); } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param inputStream + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong inputStream for SingleRow ReadBlock"); + } + + Input input = (Input) inputStream; + + int numFields = input.readInt(); + Block[] fieldBlocks = new Block[numFields]; + for (int i = 0; i < fieldBlocks.length; i++) { + fieldBlocks[i] = blockEncodingSerde.readBlock(input); + } + + return new SingleRowBlock(0, fieldBlocks); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param outputStream + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Wrong outputStream for SingleRowBlock WriteBlock"); + } + + Output output = (Output) outputStream; + SingleRowBlock singleRowBlock = (SingleRowBlock) block; + int numFields = singleRowBlock.getNumFields(); + int rowIndex = singleRowBlock.getRowIndex(); + output.writeInt(numFields); + for (int i = 0; i < numFields; i++) { + blockEncodingSerde.writeBlock(output, singleRowBlock.getRawFieldBlock(i).getRegion(rowIndex, 1)); + } + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java b/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java index 820fa235b..d95992ebc 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java @@ -13,9 +13,14 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.KryoSerializable; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.Slice; import io.airlift.slice.SliceOutput; import io.airlift.slice.Slices; +import io.prestosql.spi.PrestoException; import io.prestosql.spi.util.BloomFilter; import org.openjdk.jol.info.ClassLayout; @@ -28,6 +33,7 @@ import java.util.function.BiConsumer; import java.util.function.Function; import static io.airlift.slice.SizeOf.sizeOf; +import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR; import static io.prestosql.spi.block.BlockUtil.checkArrayRange; import static io.prestosql.spi.block.BlockUtil.checkValidRegion; import static io.prestosql.spi.block.BlockUtil.compactArray; @@ -36,18 +42,25 @@ import static io.prestosql.spi.block.BlockUtil.compactSlice; public class VariableWidthBlock extends AbstractVariableWidthBlock + implements KryoSerializable { private static final int INSTANCE_SIZE = ClassLayout.parseClass(VariableWidthBlock.class).instanceSize(); - private final int arrayOffset; - private final int positionCount; - private final Slice slice; - private final int[] offsets; + protected int arrayOffset; + private int positionCount; + private Slice slice; + protected int[] offsets; @Nullable - private final boolean[] valueIsNull; + protected boolean[] valueIsNull; - private final long retainedSizeInBytes; - private final long sizeInBytes; + private long retainedSizeInBytes; + private long sizeInBytes; + private boolean isInitialized; + + public VariableWidthBlock() + { + arrayOffset = 0; + } public VariableWidthBlock(int positionCount, Slice slice, int[] offsets, Optional valueIsNull) { @@ -82,6 +95,8 @@ public class VariableWidthBlock sizeInBytes = offsets[arrayOffset + positionCount] - offsets[arrayOffset] + ((Integer.BYTES + Byte.BYTES) * (long) positionCount); retainedSizeInBytes = INSTANCE_SIZE + slice.getRetainedSize() + sizeOf(valueIsNull) + sizeOf(offsets); + + this.isInitialized = true; } @Override @@ -297,4 +312,43 @@ public class VariableWidthBlock { return Objects.hash(arrayOffset, positionCount, slice, Arrays.hashCode(offsets), Arrays.hashCode(valueIsNull), retainedSizeInBytes, sizeInBytes); } + + @Override + public void write(Kryo kryo, Output output) + { + /* # of positions + * [Length Per Position] + * [nulls as bits] + * [buffer] + */ + output.write(getPositionCount()); + output.writeInts(offsets, arrayOffset, positionCount + 1); + output.writeBoolean(mayHaveNull()); + if (mayHaveNull()) { + output.writeBooleans(valueIsNull, 0, positionCount); + } + output.write(offsets[arrayOffset + positionCount] - offsets[arrayOffset]); + output.write(slice.byteArray(), offsets[arrayOffset], + offsets[arrayOffset + positionCount] - offsets[arrayOffset]); + } + + @Override + public void read(Kryo kryo, Input input) + { + if (isInitialized) { + throw new PrestoException(GENERIC_INTERNAL_ERROR, "Already initialized block"); + } + + positionCount = input.read(); + offsets = input.readInts(positionCount + 1); + if (input.readBoolean()) { + valueIsNull = input.readBooleans(positionCount); + } + int blockSize = input.read(); + slice = Slices.wrappedBuffer(input.readBytes(blockSize)); + + isInitialized = true; + sizeInBytes = offsets[arrayOffset + positionCount] - offsets[arrayOffset] + ((Integer.BYTES + Byte.BYTES) * (long) positionCount); + retainedSizeInBytes = INSTANCE_SIZE + slice.getRetainedSize() + sizeOf(valueIsNull) + sizeOf(offsets); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlockEncoding.java b/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlockEncoding.java index f5e4408b2..817e2cfd2 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlockEncoding.java +++ b/presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlockEncoding.java @@ -13,17 +13,25 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.Slice; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; import io.airlift.slice.Slices; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import java.io.InputStream; +import java.io.OutputStream; import static io.airlift.slice.SizeOf.SIZE_OF_INT; import static io.prestosql.spi.block.EncoderUtil.decodeNullBits; import static io.prestosql.spi.block.EncoderUtil.encodeNullsAsBits; public class VariableWidthBlockEncoding - implements BlockEncoding + extends AbstractBlockEncoding { public static final String NAME = "VARIABLE_WIDTH"; @@ -72,4 +80,70 @@ public class VariableWidthBlockEncoding return new VariableWidthBlock(0, positionCount, slice, offsets, valueIsNull); } + + @Override + public void write(Kryo kryo, Output output, VariableWidthBlock block) + { + int positionCount = block.getPositionCount(); + output.writeInt(positionCount); + output.writeInts(block.offsets, block.arrayOffset, positionCount + 1); + + output.writeBoolean(block.mayHaveNull()); + if (block.mayHaveNull()) { + output.writeBooleans(block.valueIsNull, 0, positionCount); + } + + int totalSize = block.offsets[block.arrayOffset + positionCount] - block.offsets[block.arrayOffset]; + output.writeInt(totalSize); + output.write(block.getRawSlice(0).byteArray(), block.offsets[block.arrayOffset], totalSize); + } + + @Override + public VariableWidthBlock read(Kryo kryo, Input input, Class aClass) + { + int positionCount = input.readInt(); + int[] offsets = input.readInts(positionCount + 1); + boolean[] valuesIsNull = null; + if (input.readBoolean()) { + valuesIsNull = input.readBooleans(positionCount); + } + int blockSize = input.readInt(); + Slice slice = Slices.wrappedBuffer(input.readBytes(blockSize)); + + return new VariableWidthBlock(0, positionCount, slice, offsets, valuesIsNull); + } + + /** + * Read a block from the specified input. The returned + * block should begin at the specified position. + * + * @param blockEncodingSerde + * @param input + */ + @Override + public Block readBlock(BlockEncodingSerde blockEncodingSerde, InputStream input) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(input instanceof Input)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic readblock not supported for VariableWidthBlock"); + } + + return this.read((Kryo) blockEncodingSerde.getContext(), (Input) input, VariableWidthBlock.class); + } + + /** + * Write the specified block to the specified output + * + * @param blockEncodingSerde + * @param output + * @param block + */ + @Override + public void writeBlock(BlockEncodingSerde blockEncodingSerde, OutputStream output, Block block) + { + if (!(blockEncodingSerde.getContext() instanceof Kryo) || !(output instanceof Output)) { + throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "Generic write not supported for VariableWidthBlock"); + } + + this.write((Kryo) blockEncodingSerde.getContext(), (Output) output, (VariableWidthBlock) block); + } } diff --git a/presto-spi/src/main/java/io/prestosql/spi/type/TypeSerde.java b/presto-spi/src/main/java/io/prestosql/spi/type/TypeSerde.java index 97fde21eb..86ce8c1bf 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/type/TypeSerde.java +++ b/presto-spi/src/main/java/io/prestosql/spi/type/TypeSerde.java @@ -13,6 +13,8 @@ */ package io.prestosql.spi.type; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; @@ -59,4 +61,37 @@ public final class TypeSerde output.writeInt(bytes.length); output.writeBytes(bytes); } + + public static void writeType(Output output, Type type) + { + requireNonNull(output, "output is null"); + requireNonNull(type, "type is null"); + writeLengthPrefixedString(output, type.getTypeSignature().toString()); + } + + public static Type readType(TypeManager typeManager, Input sliceInput) + { + requireNonNull(sliceInput, "sliceInput is null"); + + String name = readLengthPrefixedString(sliceInput); + Type type = typeManager.getType(parseTypeSignature(name)); + if (type == null) { + throw new IllegalArgumentException("Unknown type " + name); + } + return type; + } + + private static String readLengthPrefixedString(Input input) + { + int length = input.readInt(); + byte[] bytes = input.readBytes(length); + return new String(bytes, UTF_8); + } + + private static void writeLengthPrefixedString(Output output, String string) + { + byte[] bytes = string.getBytes(UTF_8); + output.writeInt(bytes.length); + output.writeBytes(bytes); + } } diff --git a/presto-spi/src/test/java/io/prestosql/spi/block/TestInt128ArrayBlockEncoding.java b/presto-spi/src/test/java/io/prestosql/spi/block/TestInt128ArrayBlockEncoding.java new file mode 100644 index 000000000..0a837bbfd --- /dev/null +++ b/presto-spi/src/test/java/io/prestosql/spi/block/TestInt128ArrayBlockEncoding.java @@ -0,0 +1,113 @@ +/* + * Copyright (C) 2018-2022. Huawei Technologies Co., Ltd. All rights reserved. + * 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 io.prestosql.spi.block; + +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import io.airlift.slice.InputStreamSliceInput; +import io.airlift.slice.OutputStreamSliceOutput; +import io.prestosql.spi.type.Type; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.Optional; + +import static io.prestosql.spi.block.TestingSession.SESSION; +import static io.prestosql.spi.type.BigintType.BIGINT; +import static org.testng.Assert.assertEquals; + +public class TestInt128ArrayBlockEncoding +{ + private final BlockEncodingSerde blockEncodingSerde = new TestingBlockEncodingSerde(); + private String storePath = "./target/store"; + + @BeforeClass + public void init() + { + File dir = new File(storePath); + if (!dir.exists()) { + dir.mkdirs(); + } + } + + @AfterClass + public void teardown() + { + File dir = new File(storePath); + if (dir.exists()) { + File[] files = dir.listFiles(); + for (File file : files) { + file.delete(); + } + dir.delete(); + } + } + + @Test + public void testRoundTrip() throws IOException + { + int count = 1024; + Int128ArrayBlock expectedBlock = new Int128ArrayBlock(count, Optional.empty(), getValues(count * 2)); + + OutputStreamSliceOutput sliceOutput = new OutputStreamSliceOutput(new FileOutputStream(storePath + "/" + "sliceFile.dat")); + InputStreamSliceInput sliceInput = new InputStreamSliceInput(new FileInputStream(storePath + "/" + "sliceFile.dat")); + + blockEncodingSerde.writeBlock(sliceOutput, expectedBlock); + sliceOutput.close(); + + Block actualBlock = blockEncodingSerde.readBlock(sliceInput); + sliceInput.close(); + assertBlockEquals(BIGINT, actualBlock, expectedBlock); + } + + @Test + public void testRoundTripDirect() throws FileNotFoundException + { + int count = 1024; + Int128ArrayBlock expectedBlock = new Int128ArrayBlock(count, Optional.empty(), getValues(count * 2)); + Output output = new Output(new FileOutputStream(storePath + "/" + "file.dat")); + Input input = new Input(new FileInputStream(storePath + "/" + "file.dat")); + + blockEncodingSerde.writeBlock(output, expectedBlock); + output.close(); + + Block actualBlock = blockEncodingSerde.readBlock(input); + input.close(); + + assertBlockEquals(BIGINT, actualBlock, expectedBlock); + } + + private long[] getValues(int count) + { + long[] values = new long[count]; + for (int i = 0; i < values.length; i++) { + values[i] = i; + } + return values; + } + + private static void assertBlockEquals(Type type, Block actual, Block expected) + { + for (int position = 0; position < actual.getPositionCount(); position++) { + assertEquals(type.getObjectValue(SESSION, actual, position), type.getObjectValue(SESSION, expected, position)); + } + } +} diff --git a/presto-spi/src/test/java/io/prestosql/spi/block/TestVariableWidthBlockEncoding.java b/presto-spi/src/test/java/io/prestosql/spi/block/TestVariableWidthBlockEncoding.java index 7b737abb8..3d85612c7 100644 --- a/presto-spi/src/test/java/io/prestosql/spi/block/TestVariableWidthBlockEncoding.java +++ b/presto-spi/src/test/java/io/prestosql/spi/block/TestVariableWidthBlockEncoding.java @@ -13,10 +13,25 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import com.google.common.base.Stopwatch; import io.airlift.slice.DynamicSliceOutput; +import io.airlift.slice.InputStreamSliceInput; +import io.airlift.slice.OutputStreamSliceOutput; import io.prestosql.spi.type.Type; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.concurrent.TimeUnit; + import static io.prestosql.spi.block.TestingSession.SESSION; import static io.prestosql.spi.type.VarcharType.VARCHAR; import static org.testng.Assert.assertEquals; @@ -24,6 +39,36 @@ import static org.testng.Assert.assertEquals; public class TestVariableWidthBlockEncoding { private final BlockEncodingSerde blockEncodingSerde = new TestingBlockEncodingSerde(); + private String storePath = "./target/store"; + + private Kryo kryo; + private Output output; + private Input input; + + @BeforeClass + public void init() + { + kryo = new Kryo(); + kryo.register(VariableWidthBlock.class); + + File dir = new File(storePath); + if (!dir.exists()) { + dir.mkdirs(); + } + } + + @AfterClass + public void teardown() + { + File dir = new File(storePath); + if (dir.exists()) { + File[] files = dir.listFiles(); + for (File file : files) { + file.delete(); + } + dir.delete(); + } + } @Test public void testRoundTrip() @@ -41,6 +86,86 @@ public class TestVariableWidthBlockEncoding assertBlockEquals(VARCHAR, actualBlock, expectedBlock); } + @Test + public void testRoundTripKryo() throws FileNotFoundException + { + output = new Output(new FileOutputStream(storePath + "/" + "file.dat")); + input = new Input(new FileInputStream(storePath + "/" + "file.dat")); + + BlockBuilder expectedBlockBuilder = VARCHAR.createBlockBuilder(null, 4); + VARCHAR.writeString(expectedBlockBuilder, "alice"); + VARCHAR.writeString(expectedBlockBuilder, "bob"); + VARCHAR.writeString(expectedBlockBuilder, "charlie"); + VARCHAR.writeString(expectedBlockBuilder, "dave"); + Block expectedBlock = expectedBlockBuilder.build(); + + kryo.writeObject(output, expectedBlock); + output.close(); + + Block actualBlock = kryo.readObject(input, VariableWidthBlock.class); + assertBlockEquals(VARCHAR, actualBlock, expectedBlock); + input.close(); + } + + public void testRoundTripKryoPerf100000000() throws IOException + { + int loopCount = 1000; + for (int i = 1; i <= 5; i++) { + loopCount *= 10; + loopReadWritePerfTest(loopCount); + } + } + + private void loopReadWritePerfTest(int loopCount) throws IOException + { + output = new Output(new FileOutputStream(storePath + "/" + "file.dat")); + input = new Input(new FileInputStream(storePath + "/" + "file.dat")); + + OutputStreamSliceOutput sliceOutput = new OutputStreamSliceOutput(new FileOutputStream(storePath + "/" + "sliceFile.dat")); + InputStreamSliceInput sliceInput = new InputStreamSliceInput(new FileInputStream(storePath + "/" + "sliceFile.dat")); + + BlockBuilder expectedBlockBuilder = VARCHAR.createBlockBuilder(null, 4); + VARCHAR.writeString(expectedBlockBuilder, "alice"); + VARCHAR.writeString(expectedBlockBuilder, "bob"); + VARCHAR.writeString(expectedBlockBuilder, "charlie"); + VARCHAR.writeString(expectedBlockBuilder, "dave"); + Block expectedBlock = expectedBlockBuilder.build(); + + Stopwatch watchKryoWrite = Stopwatch.createStarted(); + for (int i = 0; i < loopCount; i++) { + kryo.writeObject(output, expectedBlock); + } + watchKryoWrite.stop(); + System.out.println(String.format("[Pages: %,11d] Time to write [Kryo]: %,7d ms", loopCount, watchKryoWrite.elapsed(TimeUnit.MILLISECONDS))); + output.close(); + + Stopwatch watchSerDeWrite = Stopwatch.createStarted(); + for (int i = 0; i < loopCount; i++) { + blockEncodingSerde.writeBlock(sliceOutput, expectedBlock); + } + watchSerDeWrite.stop(); + System.out.println(String.format("[Pages: %,11d] Time to write [BlockSerDe]: %,7d ms", loopCount, watchSerDeWrite.elapsed(TimeUnit.MILLISECONDS))); + sliceOutput.close(); + + Stopwatch watchKryoRead = Stopwatch.createStarted(); + for (int i = 0; i < loopCount; i++) { + Block actualBlock = kryo.readObject(input, VariableWidthBlock.class); + assertBlockEquals(VARCHAR, actualBlock, expectedBlock); + } + watchKryoRead.stop(); + System.out.println(String.format("[Pages: %,11d] Time to read [Kryo]: %,7d ms", loopCount, watchKryoRead.elapsed(TimeUnit.MILLISECONDS))); + input.close(); + + Stopwatch watchSerDeRead = Stopwatch.createStarted(); + for (int i = 0; i < loopCount; i++) { + Block actualBlock = blockEncodingSerde.readBlock(sliceInput); + assertBlockEquals(VARCHAR, actualBlock, expectedBlock); + } + watchSerDeRead.stop(); + System.out.println(String.format("[Pages: %,11d] Time to read [BlockSerDe]: %,7d ms", loopCount, watchSerDeRead.elapsed(TimeUnit.MILLISECONDS))); + input.close(); + } + private static void assertBlockEquals(Type type, Block actual, Block expected) { for (int position = 0; position < actual.getPositionCount(); position++) { diff --git a/presto-spi/src/test/java/io/prestosql/spi/block/TestingBlockEncodingSerde.java b/presto-spi/src/test/java/io/prestosql/spi/block/TestingBlockEncodingSerde.java index cebd50423..a254bde51 100644 --- a/presto-spi/src/test/java/io/prestosql/spi/block/TestingBlockEncodingSerde.java +++ b/presto-spi/src/test/java/io/prestosql/spi/block/TestingBlockEncodingSerde.java @@ -13,14 +13,23 @@ */ package io.prestosql.spi.block; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Serializer; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; import io.airlift.slice.SliceInput; import io.airlift.slice.SliceOutput; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; +import java.io.InputStream; +import java.io.OutputStream; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import static com.google.common.base.Preconditions.checkArgument; +import static io.prestosql.spi.StandardErrorCode.TYPE_NOT_FOUND; import static java.nio.charset.StandardCharsets.UTF_8; // This class is exactly the same as BlockEncodingManager. They are in SPI and don't have access to InternalBlockEncodingSerde. @@ -28,6 +37,7 @@ public final class TestingBlockEncodingSerde implements BlockEncodingSerde { private final ConcurrentMap blockEncodings = new ConcurrentHashMap<>(); + private final Kryo kryo = new Kryo(); public TestingBlockEncodingSerde() { @@ -92,6 +102,74 @@ public final class TestingBlockEncodingSerde } } + @Override + public Block readBlock(InputStream inputStream) + { + if (!(inputStream instanceof Input)) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, + "This interface should not be called in this flow"); + } + + Input input = (Input) inputStream; + String encodingName; + // read the encoding name + encodingName = readLengthPrefixedString((Input) input); + + // look up the encoding factory + BlockEncoding blockEncoding = blockEncodings.get(encodingName); + Serializer serializer = getSerializerFromBlockEncoding(blockEncoding); + if (serializer == null) { + throw new PrestoException(TYPE_NOT_FOUND, "BlockEncoding Type not implemented: " + blockEncoding); + } + return (Block) serializer.read(kryo, input, null); + } + + @Override + public void writeBlock(OutputStream outputStream, Block block) + { + if (!(outputStream instanceof Output)) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, + "This interface should not be called in this flow"); + } + + Output output = (Output) outputStream; + + String encodingName = block.getEncodingName(); + BlockEncoding blockEncoding = blockEncodings.get(encodingName); + Serializer> serializer = getSerializerFromBlockEncoding(blockEncoding); + if (serializer == null) { + throw new PrestoException(TYPE_NOT_FOUND, "BlockEncoding Type not implemented: " + blockEncoding); + } + + // write the name to the output + writeLengthPrefixedString(output, encodingName); + + // write the block to the output + serializer.write(kryo, output, block); + } + + private Serializer> getSerializerFromBlockEncoding(BlockEncoding blockEncoding) + { + if (blockEncoding instanceof AbstractBlockEncoding) { + return (Serializer>) blockEncoding; + } + return null; + } + + private static String readLengthPrefixedString(Input input) + { + int length = input.readInt(); + byte[] bytes = input.readBytes(length); + return new String(bytes, UTF_8); + } + + private static void writeLengthPrefixedString(Output output, String value) + { + byte[] bytes = value.getBytes(UTF_8); + output.writeInt(bytes.length); + output.writeBytes(bytes); + } + private static String readLengthPrefixedString(SliceInput input) { int length = input.readInt();