   (let [tp (doto (TimePartitioning.) (.setType (-> type name .toUpperCase)))]
     (cond-> tp
       (int? expiration-ms)                   (.setExpirationMs expiration-ms)
-      (string? field)                        (.setField field)
+      (string? field)                        (.setField (clean-name field))
       (boolean? require-partition-filter)    (.setRequirePartitionFilter require-partition-filter))))
+
+
+(defn ^Clustering ->clustering
+  [fields]
+  (let [obj (Clustering.)]
+    (.setFields ^Clustering obj (mapv clean-name fields))))
 
 (defn get-bq-table-schema
   "Beware, uses bq util to get the schema!"

                            (fn [transform retrypolicy] (.withFailedInsertRetryPolicy transform retrypolicy)))}
    :time-partitioning {:docstr "Toggles write partitioning for the destination table"
                        :action (fn [transform opts]
                                  (.withTimePartitioning transform (->time-partitioning opts)))}}))

(defn custom-output-fn [cust-fn]
  (ds/sfn (fn [elt]

   [datasplash.core :as ds])
  (:import
   (com.google.api.services.bigquery.model
    TableRow TableFieldSchema TableSchema TimePartitioning
    Clustering)
   (org.apache.beam.sdk Pipeline)
   (org.apache.beam.sdk.io.gcp.bigquery
    BigQueryIO
