DictionaryEncodedColumnPartSerde: Read values using smooshReader.

The write side uses a FileSmoosher for valueWriter, which means that the
value component may be split over multiple files. In this case, the
read side needs a SmooshedFileMapper.

It is plausible for multi-valued columns to have value sections in excess
of 2 GB: imagine a segment with 10 million rows and an average of 50 values
per row. The value section would have 500 million ints, which is 2 GB.

processing/src/main/java/org/apache/druid/segment/serde/DictionaryEncodedColumnPartSerde.java

@@ -29,6 +29,7 @@
 import org.apache.druid.io.Channels;
 import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.io.smoosh.FileSmoosher;
+import org.apache.druid.java.util.common.io.smoosh.SmooshedFileMapper;
 import org.apache.druid.segment.column.BaseColumn;
 import org.apache.druid.segment.column.ColumnBuilder;
 import org.apache.druid.segment.column.ColumnConfig;
@@ -332,10 +333,10 @@ public void read(
         final WritableSupplier<ColumnarMultiInts> rMultiValuedColumn;
 
         if (hasMultipleValues) {
-          rMultiValuedColumn = readMultiValuedColumn(rVersion, buffer, rFlags);
+          rMultiValuedColumn = readMultiValuedColumn(rVersion, buffer, rFlags, builder.getFileMapper());
           rSingleValuedColumn = null;
         } else {
-          rSingleValuedColumn = readSingleValuedColumn(rVersion, buffer);
+          rSingleValuedColumn = readSingleValuedColumn(rVersion, buffer, builder.getFileMapper());
           rMultiValuedColumn = null;
         }
 
@@ -381,20 +382,29 @@ public void read(
         }
       }
 
-      private WritableSupplier<ColumnarInts> readSingleValuedColumn(VERSION version, ByteBuffer buffer)
+      private WritableSupplier<ColumnarInts> readSingleValuedColumn(
+          VERSION version,
+          ByteBuffer buffer,
+          SmooshedFileMapper smooshReader
+      )
       {
         switch (version) {
           case UNCOMPRESSED_SINGLE_VALUE:
           case UNCOMPRESSED_WITH_FLAGS:
             return VSizeColumnarInts.readFromByteBuffer(buffer);
           case COMPRESSED:
-            return CompressedVSizeColumnarIntsSupplier.fromByteBuffer(buffer, byteOrder);
+            return CompressedVSizeColumnarIntsSupplier.fromByteBuffer(buffer, byteOrder, smooshReader);
           default:
             throw new IAE("Unsupported single-value version[%s]", version);
         }
       }
 
-      private WritableSupplier<ColumnarMultiInts> readMultiValuedColumn(VERSION version, ByteBuffer buffer, int flags)
+      private WritableSupplier<ColumnarMultiInts> readMultiValuedColumn(
+          VERSION version,
+          ByteBuffer buffer,
+          int flags,
+          SmooshedFileMapper smooshReader
+      )
       {
         switch (version) {
           case UNCOMPRESSED_MULTI_VALUE: {
@@ -411,7 +421,7 @@ private WritableSupplier<ColumnarMultiInts> readMultiValuedColumn(VERSION versio
             if (Feature.MULTI_VALUE.isSet(flags)) {
               return CompressedVSizeColumnarMultiIntsSupplier.fromByteBuffer(buffer, byteOrder);
             } else if (Feature.MULTI_VALUE_V3.isSet(flags)) {
-              return V3CompressedVSizeColumnarMultiIntsSupplier.fromByteBuffer(buffer, byteOrder);
+              return V3CompressedVSizeColumnarMultiIntsSupplier.fromByteBuffer(buffer, byteOrder, smooshReader);
             } else {
               throw new IAE("Unrecognized multi-value flag[%d] for version[%s]", flags, version);
             }