Add configurable serde properties and refactor/cleanup

Author: bangtim

Diff for: lib/trino-hive-formats/pom.xml

@@ -27,12 +27,6 @@
             <artifactId>jackson-databind</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.google.code.gson</groupId>
-            <artifactId>gson</artifactId>
-            <version>2.12.1</version>
-        </dependency>
-
         <dependency>
             <groupId>com.google.errorprone</groupId>
             <artifactId>error_prone_annotations</artifactId>

Diff for: lib/trino-hive-formats/src/main/java/io/trino/hive/formats/HiveClassNames.java

@@ -21,7 +21,7 @@ public final class HiveClassNames
     public static final String COLUMNAR_SERDE_CLASS = "org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe";
     public static final String FILE_INPUT_FORMAT_CLASS = "org.apache.hadoop.mapred.FileInputFormat";
     public static final String FILE_OUTPUT_FORMAT_CLASS = "org.apache.hadoop.mapred.FileOutputFormat";
-    public static final String GROK_SERDE_CLASS = "com.amazonaws.serde.GrokSerDe";
+    public static final String GROK_SERDE_CLASS = "com.amazonaws.glue.serde.GrokSerDe";
     public static final String HIVE_IGNORE_KEY_OUTPUT_FORMAT_CLASS = "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat";
     public static final String HIVE_SEQUENCEFILE_OUTPUT_FORMAT_CLASS = "org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat";
     public static final String HUDI_PARQUET_INPUT_FORMAT = "org.apache.hudi.hadoop.HoodieParquetInputFormat";

Diff for: lib/trino-hive-formats/src/main/java/io/trino/hive/formats/line/grok/Converter.java

@@ -13,6 +13,7 @@
  */
 package io.trino.hive.formats.line.grok;
 
+import com.google.common.collect.ImmutableMap;
 import io.trino.hive.formats.line.grok.exception.GrokException;
 
 import java.text.ParseException;
@@ -36,6 +37,12 @@ private Converter() {}
 
     public static Map<String, IConverter<?>> converters = new HashMap<String, IConverter<?>>();
     public static Locale locale = Locale.ENGLISH;
+    private static final int MAX_SPEC_PARTS = 3; // field ID, datatype, and datetype arguments
+    private static final int FIELD_ID_AND_DATATYPE = 2;
+    private static final int ONLY_FIELD_ID = 1;
+    private static final int FIELD_ID_IDX = 0;
+    private static final int DATATYPE_IDX = 1;
+    private static final int DATATYPE_ARGS_IDX = 2;
 
     static {
         converters.put("byte", new ByteConverter());
@@ -60,20 +67,48 @@ private static IConverter getConverter(String key)
         return converter;
     }
 
-    public static KeyValue convert(String key, Object value, Grok grok)
+    /**
+     * Convert a value according to the specified key pattern and Grok config
+     *
+     * The key can be of the form:
+     * fieldID
+     * fieldID:datatype
+     * fieldID:datatype:datatypeArgs
+     *
+     * fieldID - Identifier of field being parsed
+     * datatype - (Optional) target data type (e.g. int, string, date)
+     * args - (Optional) arguments to the data type (e.g. date format)
+     *
+     * @param key The pattern key with components field, data type, and args
+     * @param value The value to convert
+     * @param grok Grok instance containing pattern configs and conversion settings (e.g. strict mode)
+     * @return ImmutableMap containing the field ID and its converted value
+     * @throws GrokException If conversion fails or if pattern/datatype is invalid
+     *
+     * converting a timestamp: convert("timestamp:date:yyyy-MM-dd", "2023-12-25", grok)
+     * timestamp is the field ID, date is the data type, and yyyy-MM-dd is the date format argument
+     *
+     * converting int: convert("status:int", "200", grok)
+     * status is the field ID, int is the data type
+     *
+     * using default data type from pattern: convert("message", "Hello World", grok)
+     * message is the field ID, no data type is specified, so the default data type from the pattern is used
+     *
+     */
+    public static ImmutableMap<String, Object> convert(String key, Object value, Grok grok)
             throws GrokException
     {
-        String[] spec = key.split(";|:", 3);
+        String[] spec = key.split(";|:", MAX_SPEC_PARTS);
         try {
             // process situations with field id [and datatype]
-            if (spec.length <= 2) {
+            if (spec.length <= FIELD_ID_AND_DATATYPE) {
                 String pattern = grok.getGrokPatternPatterns().get(key); // actual pattern name
                 String defaultDataType = grok.getGrokPatternDefaultDatatype().get(pattern); // default datatype of the pattern
                 // process Date datatype with no format arguments
                 // 1. not in strict mode && no assigned data type && the default data type is datetime or date
                 // 2. assigned data type is datetime or date && no date format argument
-                if ((!grok.getStrictMode() && spec.length == 1 && defaultDataType != null && (defaultDataType.equals("datetime") || defaultDataType.equals("date")))
-                        || (spec.length == 2 && (spec[1].equals("datetime") || spec[1].equals("date")))) {
+                if ((!grok.getStrictMode() && spec.length == ONLY_FIELD_ID && defaultDataType != null && (defaultDataType.equals("datetime") || defaultDataType.equals("date")))
+                        || (spec.length == FIELD_ID_AND_DATATYPE && (spec[DATATYPE_IDX].equals("datetime") || spec[DATATYPE_IDX].equals("date")))) {
                     // check whether to get the date format already when parsing the previous records
                     String dateFormat = grok.getGrokPatternPatterns().get(key + "dateformat");
                     Date date = null;
@@ -100,30 +135,30 @@ public static KeyValue convert(String key, Object value, Grok grok)
                     }
                     if (date != null) {
                         // if parse successfully, return date object
-                        return new KeyValue(spec[0], date);
+                        return ImmutableMap.of(spec[FIELD_ID_IDX], date);
                     }
                     else {
                         // if failed, return string object
-                        return new KeyValue(spec[0], String.valueOf(value));
+                        return ImmutableMap.of(spec[FIELD_ID_IDX], String.valueOf(value));
                     }
                 }
-                else if (spec.length == 1) {
+                else if (spec.length == ONLY_FIELD_ID) {
                     if (grok.getStrictMode()) {
                         // if in strict mode, never do automatic data type conversion
                         defaultDataType = null;
                     }
                     // process situations with only field id (check default datatype, except date and datetime)
-                    return new KeyValue(spec[0],
+                    return ImmutableMap.of(spec[FIELD_ID_IDX],
                             defaultDataType == null ? String.valueOf(value) : getConverter(defaultDataType).convert(String.valueOf(value)));
                 }
                 else {
                     // process situations with field id and datatype (except date and datetime)
-                    return new KeyValue(spec[0], getConverter(spec[1]).convert(String.valueOf(value)));
+                    return ImmutableMap.of(spec[FIELD_ID_IDX], getConverter(spec[DATATYPE_IDX]).convert(String.valueOf(value)));
                 }
             }
-            else if (spec.length == 3) {
+            else if (spec.length == MAX_SPEC_PARTS) {
                 // process situations with field id, datatype and datatype arguments
-                return new KeyValue(spec[0], getConverter(spec[1]).convert(String.valueOf(value), spec[2]));
+                return ImmutableMap.of(spec[FIELD_ID_IDX], getConverter(spec[DATATYPE_IDX]).convert(String.valueOf(value), spec[DATATYPE_ARGS_IDX]));
             }
             else {
                 throw new GrokException("Unsupported spec : " + key);
@@ -132,11 +167,11 @@ else if (spec.length == 3) {
         catch (Exception e) {
             if (!grok.getStrictMode()) {
                 // if not in strict mode, try to convert everything to string when meeting a data type conversion error
-                return new KeyValue(spec[0], String.valueOf(value));
+                return ImmutableMap.of(spec[0], String.valueOf(value));
             }
             else {
                 // if in strict mode, throw exception when meeting a data type conversion error
-                throw new GrokException("Unable to finish data type conversion of " + spec[0] + ":" + e.getMessage());
+                throw new GrokException("Unable to finish data type conversion of " + spec[FIELD_ID_IDX] + ":" + e.getMessage());
             }
         }
     }

Diff for: lib/trino-hive-formats/src/main/java/io/trino/hive/formats/line/grok/Garbage.java

@@ -61,38 +61,6 @@ public void addToRename(String origin, Object value)
         }
     }
 
-    /**
-     * Set a field to be removed when exporting the final output.
-     *
-     * @param name of the field to remove
-     */
-    public void addToRemove(String name)
-    {
-        if (name == null) {
-            return;
-        }
-
-        if (!name.isEmpty()) {
-            toRemove.add(name);
-        }
-    }
-
-    /**
-     * Set a list of field name to be removed when exporting the final output.
-     *
-     * @param lst list of elem to remove
-     */
-    public void addToRemove(List<String> lst)
-    {
-        if (lst == null) {
-            return;
-        }
-
-        if (!lst.isEmpty()) {
-            toRemove.addAll(lst);
-        }
-    }
-
     /**
      * Remove from the map the unwilling items.
      *