This is an automated email from the ASF dual-hosted git repository.
gian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new b645d09c5dc move long and double nested field serialization to later
phase of serialization (#16769)
b645d09c5dc is described below
commit b645d09c5dc5791e538532efbd44e203e005a9fb
Author: Clint Wylie <cwy...@apache.org>
AuthorDate: Mon Jul 22 21:14:30 2024 -0700
move long and double nested field serialization to later phase of
serialization (#16769)
changes:
* moves value column serializer initialization, call to `writeValue` method
to `GlobalDictionaryEncodedFieldColumnWriter.writeTo` instead of during
`GlobalDictionaryEncodedFieldColumnWriter.addValue`. This shift means these
numeric value columns are now done in the per field section that happens after
serializing the nested column raw data, so only a single compression buffer and
temp file will be needed at a time instead of the total number of nested
literal fields present in the col [...]
---
.../druid/segment/nested/DictionaryIdLookup.java | 158 +++++++++++++--------
.../GlobalDictionaryEncodedFieldColumnWriter.java | 10 +-
.../nested/ScalarDoubleFieldColumnWriter.java | 26 ++--
.../nested/ScalarLongFieldColumnWriter.java | 26 ++--
4 files changed, 128 insertions(+), 92 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/DictionaryIdLookup.java
b/processing/src/main/java/org/apache/druid/segment/nested/DictionaryIdLookup.java
index f4176db220c..6827497f7a6 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/DictionaryIdLookup.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/DictionaryIdLookup.java
@@ -99,42 +99,27 @@ public final class DictionaryIdLookup implements Closeable
this.arrayDictionaryWriter = arrayDictionaryWriter;
}
- public int lookupString(@Nullable String value)
+ @Nullable
+ public Object getDictionaryValue(int id)
{
- if (stringDictionary == null) {
- // GenericIndexed v2 can write to multiple files if the dictionary is
larger than 2gb, so we use a smooshfile
- // for strings because of this. if other type dictionary writers could
potentially use multiple internal files
- // in the future, we should transition them to using this approach as
well (or build a combination smoosher and
- // mapper so that we can have a mutable smoosh)
- File stringSmoosh = FileUtils.createTempDirInLocation(tempBasePath,
StringUtils.urlEncode(name) + "__stringTempSmoosh");
- stringDictionaryFile = stringSmoosh.toPath();
- final String fileName =
NestedCommonFormatColumnSerializer.getInternalFileName(
- name,
- NestedCommonFormatColumnSerializer.STRING_DICTIONARY_FILE_NAME
- );
-
- try (
- final FileSmoosher smoosher = new FileSmoosher(stringSmoosh);
- final SmooshedWriter writer = smoosher.addWithSmooshedWriter(
- fileName,
- stringDictionaryWriter.getSerializedSize()
- )
- ) {
- stringDictionaryWriter.writeTo(writer, smoosher);
- writer.close();
- smoosher.close();
- stringBufferMapper = SmooshedFileMapper.load(stringSmoosh);
- final ByteBuffer stringBuffer = stringBufferMapper.mapFile(fileName);
- stringDictionary =
StringEncodingStrategies.getStringDictionarySupplier(
- stringBufferMapper,
- stringBuffer,
- ByteOrder.nativeOrder()
- ).get();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
+ ensureStringDictionaryLoaded();
+ ensureLongDictionaryLoaded();
+ ensureDoubleDictionaryLoaded();
+ ensureArrayDictionaryLoaded();
+ if (id < longOffset()) {
+ return StringUtils.fromUtf8Nullable(stringDictionary.get(id));
+ } else if (id < doubleOffset()) {
+ return longDictionary.get(id - longOffset());
+ } else if (id < arrayOffset()) {
+ return doubleDictionary.get(id - doubleOffset());
+ } else {
+ return arrayDictionary.get(id - arrayOffset());
}
+ }
+
+ public int lookupString(@Nullable String value)
+ {
+ ensureStringDictionaryLoaded();
final byte[] bytes = StringUtils.toUtf8Nullable(value);
final int index = stringDictionary.indexOf(bytes == null ? null :
ByteBuffer.wrap(bytes));
if (index < 0) {
@@ -145,13 +130,7 @@ public final class DictionaryIdLookup implements Closeable
public int lookupLong(@Nullable Long value)
{
- if (longDictionary == null) {
- longDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.LONG_DICTIONARY_FILE_NAME);
- longBuffer = mapWriter(longDictionaryFile, longDictionaryWriter);
- longDictionary = FixedIndexed.read(longBuffer, TypeStrategies.LONG,
ByteOrder.nativeOrder(), Long.BYTES).get();
- // reset position
- longBuffer.position(0);
- }
+ ensureLongDictionaryLoaded();
final int index = longDictionary.indexOf(value);
if (index < 0) {
throw DruidException.defensive("Value not found in column[%s] long
dictionary", name);
@@ -161,18 +140,7 @@ public final class DictionaryIdLookup implements Closeable
public int lookupDouble(@Nullable Double value)
{
- if (doubleDictionary == null) {
- doubleDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.DOUBLE_DICTIONARY_FILE_NAME);
- doubleBuffer = mapWriter(doubleDictionaryFile, doubleDictionaryWriter);
- doubleDictionary = FixedIndexed.read(
- doubleBuffer,
- TypeStrategies.DOUBLE,
- ByteOrder.nativeOrder(),
- Double.BYTES
- ).get();
- // reset position
- doubleBuffer.position(0);
- }
+ ensureDoubleDictionaryLoaded();
final int index = doubleDictionary.indexOf(value);
if (index < 0) {
throw DruidException.defensive("Value not found in column[%s] double
dictionary", name);
@@ -182,13 +150,7 @@ public final class DictionaryIdLookup implements Closeable
public int lookupArray(@Nullable int[] value)
{
- if (arrayDictionary == null) {
- arrayDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.ARRAY_DICTIONARY_FILE_NAME);
- arrayBuffer = mapWriter(arrayDictionaryFile, arrayDictionaryWriter);
- arrayDictionary = FrontCodedIntArrayIndexed.read(arrayBuffer,
ByteOrder.nativeOrder()).get();
- // reset position
- arrayBuffer.position(0);
- }
+ ensureArrayDictionaryLoaded();
final int index = arrayDictionary.indexOf(value);
if (index < 0) {
throw DruidException.defensive("Value not found in column[%s] array
dictionary", name);
@@ -256,6 +218,82 @@ public final class DictionaryIdLookup implements Closeable
return doubleOffset() + (doubleDictionaryWriter != null ?
doubleDictionaryWriter.getCardinality() : 0);
}
+ private void ensureStringDictionaryLoaded()
+ {
+ if (stringDictionary == null) {
+ // GenericIndexed v2 can write to multiple files if the dictionary is
larger than 2gb, so we use a smooshfile
+ // for strings because of this. if other type dictionary writers could
potentially use multiple internal files
+ // in the future, we should transition them to using this approach as
well (or build a combination smoosher and
+ // mapper so that we can have a mutable smoosh)
+ File stringSmoosh = FileUtils.createTempDirInLocation(tempBasePath,
StringUtils.urlEncode(name) + "__stringTempSmoosh");
+ stringDictionaryFile = stringSmoosh.toPath();
+ final String fileName =
NestedCommonFormatColumnSerializer.getInternalFileName(
+ name,
+ NestedCommonFormatColumnSerializer.STRING_DICTIONARY_FILE_NAME
+ );
+
+ try (
+ final FileSmoosher smoosher = new FileSmoosher(stringSmoosh);
+ final SmooshedWriter writer = smoosher.addWithSmooshedWriter(
+ fileName,
+ stringDictionaryWriter.getSerializedSize()
+ )
+ ) {
+ stringDictionaryWriter.writeTo(writer, smoosher);
+ writer.close();
+ smoosher.close();
+ stringBufferMapper = SmooshedFileMapper.load(stringSmoosh);
+ final ByteBuffer stringBuffer = stringBufferMapper.mapFile(fileName);
+ stringDictionary =
StringEncodingStrategies.getStringDictionarySupplier(
+ stringBufferMapper,
+ stringBuffer,
+ ByteOrder.nativeOrder()
+ ).get();
+ }
+ catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void ensureLongDictionaryLoaded()
+ {
+ if (longDictionary == null) {
+ longDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.LONG_DICTIONARY_FILE_NAME);
+ longBuffer = mapWriter(longDictionaryFile, longDictionaryWriter);
+ longDictionary = FixedIndexed.read(longBuffer, TypeStrategies.LONG,
ByteOrder.nativeOrder(), Long.BYTES).get();
+ // reset position
+ longBuffer.position(0);
+ }
+ }
+
+ private void ensureDoubleDictionaryLoaded()
+ {
+ if (doubleDictionary == null) {
+ doubleDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.DOUBLE_DICTIONARY_FILE_NAME);
+ doubleBuffer = mapWriter(doubleDictionaryFile, doubleDictionaryWriter);
+ doubleDictionary = FixedIndexed.read(
+ doubleBuffer,
+ TypeStrategies.DOUBLE,
+ ByteOrder.nativeOrder(),
+ Double.BYTES
+ ).get();
+ // reset position
+ doubleBuffer.position(0);
+ }
+ }
+
+ private void ensureArrayDictionaryLoaded()
+ {
+ if (arrayDictionary == null && arrayDictionaryWriter != null) {
+ arrayDictionaryFile = makeTempFile(name +
NestedCommonFormatColumnSerializer.ARRAY_DICTIONARY_FILE_NAME);
+ arrayBuffer = mapWriter(arrayDictionaryFile, arrayDictionaryWriter);
+ arrayDictionary = FrontCodedIntArrayIndexed.read(arrayBuffer,
ByteOrder.nativeOrder()).get();
+ // reset position
+ arrayBuffer.position(0);
+ }
+ }
+
private Path makeTempFile(String name)
{
try {
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/GlobalDictionaryEncodedFieldColumnWriter.java
b/processing/src/main/java/org/apache/druid/segment/nested/GlobalDictionaryEncodedFieldColumnWriter.java
index aa6a71ae754..d9f00bb2321 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/GlobalDictionaryEncodedFieldColumnWriter.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/GlobalDictionaryEncodedFieldColumnWriter.java
@@ -117,8 +117,8 @@ public abstract class
GlobalDictionaryEncodedFieldColumnWriter<T>
}
/**
- * Hook to allow implementors the chance to do additional operations during
{@link #addValue(int, Object)}, such as
- * writing an additional value column
+ * Hook to allow implementors the chance to do additional operations during
{@link #writeTo(int, FileSmoosher)}, such
+ * as writing an additional value column
*/
void writeValue(@Nullable T value) throws IOException
{
@@ -159,7 +159,6 @@ public abstract class
GlobalDictionaryEncodedFieldColumnWriter<T>
localId = localDictionary.add(globalId);
}
intermediateValueWriter.write(localId);
- writeValue(value);
cursorPosition++;
}
@@ -168,11 +167,9 @@ public abstract class
GlobalDictionaryEncodedFieldColumnWriter<T>
*/
private void fillNull(int row) throws IOException
{
- final T value = processValue(row, null);
final int localId = localDictionary.add(0);
while (cursorPosition < row) {
intermediateValueWriter.write(localId);
- writeValue(value);
cursorPosition++;
}
}
@@ -252,6 +249,7 @@ public abstract class
GlobalDictionaryEncodedFieldColumnWriter<T>
final int unsortedLocalId = rows.nextInt();
final int sortedLocalId = unsortedToSorted[unsortedLocalId];
encodedValueSerializer.addValue(sortedLocalId);
+ writeValue((T)
globalDictionaryIdLookup.getDictionaryValue(unsortedToGlobal[unsortedLocalId]));
bitmaps[sortedLocalId].add(rowCount++);
}
@@ -307,7 +305,7 @@ public abstract class
GlobalDictionaryEncodedFieldColumnWriter<T>
}
}
- private void openColumnSerializer(SegmentWriteOutMedium medium, int maxId)
throws IOException
+ public void openColumnSerializer(SegmentWriteOutMedium medium, int maxId)
throws IOException
{
if (indexSpec.getDimensionCompression() !=
CompressionStrategy.UNCOMPRESSED) {
this.version = DictionaryEncodedColumnPartSerde.VERSION.COMPRESSED;
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleFieldColumnWriter.java
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleFieldColumnWriter.java
index 8ccd528715b..09e8dc121c8 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleFieldColumnWriter.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleFieldColumnWriter.java
@@ -59,12 +59,22 @@ public final class ScalarDoubleFieldColumnWriter extends
GlobalDictionaryEncoded
}
@Override
- public void open() throws IOException
+ void writeValue(@Nullable Double value) throws IOException
+ {
+ if (value == null) {
+ doublesSerializer.add(0.0);
+ } else {
+ doublesSerializer.add(value);
+ }
+ }
+
+ @Override
+ public void openColumnSerializer(SegmentWriteOutMedium medium, int maxId)
throws IOException
{
- super.open();
+ super.openColumnSerializer(medium, maxId);
doublesSerializer = CompressionFactory.getDoubleSerializer(
fieldName,
- segmentWriteOutMedium,
+ medium,
StringUtils.format("%s.double_column", fieldName),
ByteOrder.nativeOrder(),
indexSpec.getDimensionCompression(),
@@ -73,16 +83,6 @@ public final class ScalarDoubleFieldColumnWriter extends
GlobalDictionaryEncoded
doublesSerializer.open();
}
- @Override
- void writeValue(@Nullable Double value) throws IOException
- {
- if (value == null) {
- doublesSerializer.add(0.0);
- } else {
- doublesSerializer.add(value);
- }
- }
-
@Override
void writeColumnTo(WritableByteChannel channel, FileSmoosher smoosher)
throws IOException
{
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongFieldColumnWriter.java
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongFieldColumnWriter.java
index 66b5eca18d9..d9191c4e805 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongFieldColumnWriter.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongFieldColumnWriter.java
@@ -59,12 +59,22 @@ public final class ScalarLongFieldColumnWriter extends
GlobalDictionaryEncodedFi
}
@Override
- public void open() throws IOException
+ void writeValue(@Nullable Long value) throws IOException
+ {
+ if (value == null) {
+ longsSerializer.add(0L);
+ } else {
+ longsSerializer.add(value);
+ }
+ }
+
+ @Override
+ public void openColumnSerializer(SegmentWriteOutMedium medium, int maxId)
throws IOException
{
- super.open();
+ super.openColumnSerializer(medium, maxId);
longsSerializer = CompressionFactory.getLongSerializer(
fieldName,
- segmentWriteOutMedium,
+ medium,
StringUtils.format("%s.long_column", fieldName),
ByteOrder.nativeOrder(),
indexSpec.getLongEncoding(),
@@ -74,16 +84,6 @@ public final class ScalarLongFieldColumnWriter extends
GlobalDictionaryEncodedFi
longsSerializer.open();
}
- @Override
- void writeValue(@Nullable Long value) throws IOException
- {
- if (value == null) {
- longsSerializer.add(0L);
- } else {
- longsSerializer.add(value);
- }
- }
-
@Override
void writeColumnTo(WritableByteChannel channel, FileSmoosher smoosher)
throws IOException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
