HIVE-29866: Optimize LLAP IO cache encoding path for text tables with MultiDelimitSerDe

llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/VectorDeserializeOrcWriter.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
@@ -55,6 +56,7 @@
 import org.apache.hadoop.hive.ql.plan.PartitionDesc;
 import org.apache.hadoop.hive.serde.serdeConstants;
 import org.apache.hadoop.hive.serde2.Deserializer;
+import org.apache.hadoop.hive.serde2.MultiDelimitSerDe;
 import org.apache.hadoop.hive.serde2.SerDeException;
 import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
 import org.apache.hadoop.hive.serde2.lazy.LazySerDeParameters;
@@ -103,9 +105,16 @@
       List<Integer> sourceIncludes, boolean[] cacheIncludes, int allocSize, ExecutorService encodeExecutor)
       throws IOException {
     // Vector SerDe can be disabled both on client and server side.
+    // MultiDelimitSerDe is accepted alongside LazySimpleSerDe: it's the same
+    // text-row shape, only its top-level FIELD_DELIM can span multiple bytes.
+    // The reader (LazySimpleDeserializeRead) picks the multi-byte scanner
+    // when LazySerDeParameters.setFieldDelimMulti has been called below —
+    // otherwise the specialized single-byte hot loop runs unchanged.
+    final boolean isLazySimple = serDe instanceof LazySimpleSerDe;
+    final boolean isMultiDelim = serDe instanceof MultiDelimitSerDe;
     if (!HiveConf.getBoolVar(daemonConf, ConfVars.LLAP_IO_ENCODE_VECTOR_SERDE_ENABLED)
         || !HiveConf.getBoolVar(jobConf, ConfVars.LLAP_IO_ENCODE_VECTOR_SERDE_ENABLED)
-        || !(sourceIf instanceof TextInputFormat) || !(serDe instanceof LazySimpleSerDe)) {
+        || !(sourceIf instanceof TextInputFormat) || !(isLazySimple || isMultiDelim)) {
       return new DeserializerOrcWriter(serDe, sourceOi, allocSize);
     }
     Path path = splitPath.getFileSystem(jobConf).makeQualified(splitPath);
@@ -121,6 +130,16 @@
         + serdeConstants.SERIALIZATION_LAST_COLUMN_TAKES_REST);
       return new DeserializerOrcWriter(serDe, sourceOi, allocSize);
     }
+    // MultiDelimitSerDe's own row parser (LazyStruct.parseMultiDelimit) does
+    // NOT honour escape.delim for the top-level delimiter. Combining a
+    // multi-byte field delim with an escape char in the fast path would let
+    // it silently diverge from the slow path, so bail to DeserializerOrcWriter
+    // whenever both are configured on a MultiDelimit table.
+    if (isMultiDelim && tblProps.getProperty(serdeConstants.ESCAPE_CHAR) != null) {
+      LlapIoImpl.LOG.info("Not using VectorDeserializeOrcWriter: MultiDelimitSerDe with "
+          + serdeConstants.ESCAPE_CHAR + " is not supported by the vectorized fast path");
+      return new DeserializerOrcWriter(serDe, sourceOi, allocSize);
+    }
     for (StructField sf : sourceOi.getAllStructFieldRefs()) {
       Category c = sf.getFieldObjectInspector().getCategory();
       if (c != Category.PRIMITIVE) {
@@ -130,11 +149,13 @@
     }
     LlapIoImpl.LOG.info("Creating VertorDeserializeOrcWriter for " + path);
     return new VectorDeserializeOrcWriter(
-        jobConf, tblProps, sourceOi, sourceIncludes, cacheIncludes, allocSize, encodeExecutor);
+        jobConf, tblProps, sourceOi, sourceIncludes, cacheIncludes, allocSize, encodeExecutor,
+        isMultiDelim);
   }
 
   private VectorDeserializeOrcWriter(Configuration conf, Properties tblProps, StructObjectInspector sourceOi,
-      List<Integer> sourceIncludes, boolean[] cacheIncludes, int allocSize, ExecutorService encodeExecutor)
+      List<Integer> sourceIncludes, boolean[] cacheIncludes, int allocSize, ExecutorService encodeExecutor,
+      boolean isMultiDelim)
       throws IOException {
     super(sourceOi, allocSize);
     // See also: the usage of VectorDeserializeType, for binary. For now, we only want text.
@@ -143,7 +164,8 @@
     this.cacheIncludes = cacheIncludes;
     this.sourceBatch = vrbCtx.createVectorizedRowBatch();
     deserializeRead = new LazySimpleDeserializeRead(vrbCtx.getRowColumnTypeInfos(),
-      vrbCtx.getRowdataTypePhysicalVariations(),/* useExternalBuffer */ true, createSerdeParams(conf, tblProps));
+      vrbCtx.getRowdataTypePhysicalVariations(),/* useExternalBuffer */ true,
+      createSerdeParams(conf, tblProps, isMultiDelim));
     vectorDeserializeRow = new VectorDeserializeRow<LazySimpleDeserializeRead>(deserializeRead);
     int colCount = vrbCtx.getRowColumnTypeInfos().length;
     boolean[] includes = null;
@@ -212,8 +234,10 @@
       .HIVE_VECTORIZED_INPUT_FORMAT_SUPPORTS_ENABLED).equalsIgnoreCase("decimal_64");
     final String serde = tblProps.getProperty(serdeConstants.SERIALIZATION_LIB);
     final String inputFormat = tblProps.getProperty(hive_metastoreConstants.FILE_INPUT_FORMAT);
-    final boolean isTextFormat = inputFormat != null && inputFormat.equals(TextInputFormat.class.getName()) &&
-      serde != null && serde.equals(LazySimpleSerDe.class.getName());
+    final boolean isTextFormat = inputFormat != null && inputFormat.equals(TextInputFormat.class.getName())
+      && serde != null
+      && (serde.equals(LazySimpleSerDe.class.getName())
+          || serde.equals(MultiDelimitSerDe.class.getName()));
     List<DataTypePhysicalVariation> dataTypePhysicalVariations = new ArrayList<>();
     if (isTextFormat) {
       StructTypeInfo structTypeInfo = (StructTypeInfo) TypeInfoUtils.getTypeInfoFromObjectInspector(oi);
@@ -245,9 +269,22 @@
   }
 
   private static LazySerDeParameters createSerdeParams(
-      Configuration conf, Properties tblProps) throws IOException {
+      Configuration conf, Properties tblProps, boolean isMultiDelim) throws IOException {
     try {
-      return new LazySerDeParameters(conf, tblProps, LazySimpleSerDe.class.getName());
+      LazySerDeParameters params =
+          new LazySerDeParameters(conf, tblProps, LazySimpleSerDe.class.getName());
+      // For MultiDelimitSerDe tables, if the field delimiter is > 1 byte, opt
+      // the reader into its multi-byte scan branch. LazySerDeParameters
+      // silently ignores single-byte values here, so misconfigured MultiDelim
+      // tables (e.g. FIELD_DELIM=",") still take the specialized single-byte
+      // fast path via separators[0].
+      if (isMultiDelim) {
+        String rawDelim = tblProps.getProperty(serdeConstants.FIELD_DELIM);
+        if (rawDelim != null && rawDelim.length() > 1) {
+          params.setFieldDelimMulti(rawDelim.getBytes(StandardCharsets.UTF_8));
+        }
+      }
+      return params;
     } catch (SerDeException e) {
       throw new IOException(e);
     }

llap-server/src/test/org/apache/hadoop/hive/llap/io/encoded/TestVectorDeserializeOrcWriter.java

@@ -19,24 +19,32 @@
 package org.apache.hadoop.hive.llap.io.encoded;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Properties;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.common.io.encoded.EncodedColumnBatch;
 import org.apache.hadoop.hive.llap.io.api.impl.ColumnVectorBatch;
 import org.apache.hadoop.hive.llap.io.decode.EncodedDataConsumer;
 import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
 import org.apache.hadoop.hive.ql.io.orc.WriterImpl;
 import org.apache.hadoop.hive.ql.io.orc.encoded.Consumer;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.hive.serde2.lazy.LazySerDeParameters;
 import org.apache.hive.common.util.FixedSizedObjectPool;
 import org.apache.orc.impl.SchemaEvolution;
 
 import org.junit.Test;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.CALLS_REAL_METHODS;
 import static org.mockito.Mockito.mock;
@@ -145,6 +153,71 @@ private static VectorDeserializeOrcWriter createOrcWriter(
     return orcWriter;
   }
 
+  // --- createSerdeParams: MultiDelimit routing → fieldDelimMulti wiring -----
+
+  private static LazySerDeParameters invokeCreateSerdeParams(Properties tblProps,
+      boolean isMultiDelim) throws Exception {
+    Method m = VectorDeserializeOrcWriter.class.getDeclaredMethod(
+        "createSerdeParams", Configuration.class, Properties.class, boolean.class);
+    m.setAccessible(true);
+    return (LazySerDeParameters) m.invoke(null, new Configuration(false), tblProps, isMultiDelim);
+  }
+
+  /**
+   * When routing sees a MultiDelimitSerDe with a multi-byte field.delim, the
+   * LazySerDeParameters we hand to LazySimpleDeserializeRead must carry the
+   * raw delimiter bytes — that's the signal the reader uses to switch to its
+   * multi-byte scan branch (and hence keep the row on the vectorized LLAP
+   * fast path instead of falling back to DeserializerOrcWriter).
+   */
+  @Test
+  public void testCreateSerdeParamsMultiDelimSetsFieldDelimMulti() throws Exception {
+    Properties tblProps = new Properties();
+    tblProps.setProperty(serdeConstants.FIELD_DELIM, "~|");
+    tblProps.setProperty(serdeConstants.SERIALIZATION_FORMAT, "~|");
+    tblProps.setProperty(serdeConstants.LIST_COLUMNS, "a,b");
+    tblProps.setProperty(serdeConstants.LIST_COLUMN_TYPES, "string:string");
+
+    LazySerDeParameters params = invokeCreateSerdeParams(tblProps, /*isMultiDelim*/ true);
+    assertArrayEquals("~|".getBytes(StandardCharsets.UTF_8), params.getFieldDelimMulti());
+  }
+
+  /**
+   * A LazySimpleSerDe table (isMultiDelim=false) must NEVER opt into the
+   * multi-byte scan branch, even if someone shoved a >1-byte value into
+   * field.delim — the slow path silently truncates to a single byte, and
+   * diverging here would produce different results between fast and slow
+   * paths for the same table.
+   */
+  @Test
+  public void testCreateSerdeParamsLazySimpleNeverSetsFieldDelimMulti() throws Exception {
+    Properties tblProps = new Properties();
+    tblProps.setProperty(serdeConstants.FIELD_DELIM, "~|");
+    tblProps.setProperty(serdeConstants.SERIALIZATION_FORMAT, "~|");
+    tblProps.setProperty(serdeConstants.LIST_COLUMNS, "a,b");
+    tblProps.setProperty(serdeConstants.LIST_COLUMN_TYPES, "string:string");
+
+    LazySerDeParameters params = invokeCreateSerdeParams(tblProps, /*isMultiDelim*/ false);
+    assertNull(params.getFieldDelimMulti());
+  }
+
+  /**
+   * A MultiDelimit table whose field.delim is only one byte still takes the
+   * specialized single-byte fast path — setFieldDelimMulti drops sub-2-byte
+   * values, so separators[0] remains the source of truth.
+   */
+  @Test
+  public void testCreateSerdeParamsMultiDelimWithSingleByteFallsThrough() throws Exception {
+    Properties tblProps = new Properties();
+    tblProps.setProperty(serdeConstants.FIELD_DELIM, "|");
+    tblProps.setProperty(serdeConstants.SERIALIZATION_FORMAT, "|");
+    tblProps.setProperty(serdeConstants.LIST_COLUMNS, "a,b");
+    tblProps.setProperty(serdeConstants.LIST_COLUMN_TYPES, "string:string");
+
+    LazySerDeParameters params = invokeCreateSerdeParams(tblProps, /*isMultiDelim*/ true);
+    assertNull(params.getFieldDelimMulti());
+  }
+
   private static EncodedDataConsumer createBlankEncodedDataConsumer() {
     return new EncodedDataConsumer(null, 1, null, null) {
       @Override

serde/src/java/org/apache/hadoop/hive/serde2/lazy/LazySerDeParameters.java

@@ -60,11 +60,18 @@ public class LazySerDeParameters implements LazyObjectInspectorParameters {
   private Properties tableProperties;
   private String serdeName;
 
-  // The list of bytes used for the separators in the column (a nested struct 
+  // The list of bytes used for the separators in the column (a nested struct
   // such as Array<Array<int>> will use multiple separators).
   // The list of separators + escapeChar are the bytes required to be escaped.
   private byte[] separators;
 
+  // Populated only when the top-level field delimiter is longer than one byte
+  // (i.e. by MultiDelimitSerDe). Never populated for LazySimpleSerDe callers,
+  // so existing single-byte behaviour is preserved. The single-byte `separators[0]`
+  // slot is still set to the first byte of the multi-byte delim so that
+  // needsEscape[] and any legacy consumers of getSeparators() keep working.
+  private byte[] fieldDelimMulti;
+
   private Text nullSequence;
 
   private TypeInfo rowTypeInfo;
@@ -196,6 +203,31 @@ public byte[] getSeparators() {
     return separators;
   }
 
+  /**
+   * @return the raw (potentially multi-byte) top-level field delimiter, or
+   *         {@code null} when the effective delimiter is a single byte (the
+   *         value in {@code separators[0]} is authoritative in that case).
+   *
+   * Set by callers that know they're wiring up a multi-byte delimiter
+   * (e.g. LLAP's {@code VectorDeserializeOrcWriter} when the SerDe is
+   * {@code MultiDelimitSerDe}). Left {@code null} for plain LazySimpleSerDe so
+   * its single-byte fast path is preserved verbatim.
+   */
+  public byte[] getFieldDelimMulti() {
+    return fieldDelimMulti;
+  }
+
+  /**
+   * Opt-in switch for the multi-byte top-level delimiter. Only meaningful when
+   * {@code delim.length > 1}; shorter values are ignored (the {@code separators[0]}
+   * byte remains the source of truth). Never set this for a
+   * LazySimpleSerDe-backed table — its slow path still truncates to a single
+   * byte, so opting the fast path into multi-byte would silently diverge.
+   */
+  public void setFieldDelimMulti(byte[] delim) {
+    this.fieldDelimMulti = (delim != null && delim.length > 1) ? delim : null;
+  }
+
   public Text getNullSequence() {
     return nullSequence;
   }