feat(operator): read a timestamp cast over the years the engine reaches

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala

@@ -137,10 +137,43 @@ trait StandaloneCodeGenerator {
     * to start.
     */
   def standaloneImports(): Seq[String] = Seq.empty
+
+  /**
+    * A helper each input frame goes through before this operator's code runs,
+    * for an operator the engine hands its table in a different form than the
+    * script's upstream does. The translator defines it once and feeds the
+    * operator what it returns, leaving the upstream frame as it was.
+    *
+    * A Python operator runs in pyamber, which hands it a DOUBLE column as
+    * float64, where a NaN and a null are one value. A source can hand the script
+    * a Float64 column that keeps them apart, and an operator written against
+    * float64 then answers differently.
+    */
+  def standaloneInputAdapter(): Option[StandaloneCodeGenerator.InputAdapter] =
+    this match {
+      case _: PythonOperatorDescriptor => Some(StandaloneCodeGenerator.PythonOperatorInput)
+      case _                           => None
+    }
 }
 
 object StandaloneCodeGenerator {
 
+  /** A helper's name and the definition that binds it. */
+  final case class InputAdapter(name: String, definition: String)
+
+  /** `_texera_python_operator_input`: a copy with Float64 columns as float64, the way pyamber
+    * hands them over, so the upstream frame another operator reads stays as it was.
+    */
+  val PythonOperatorInput: InputAdapter = InputAdapter(
+    "_texera_python_operator_input",
+    """def _texera_python_operator_input(df):
+      |    df = df.copy()
+      |    for _c in df.columns:
+      |        if isinstance(df[_c].dtype, (pd.Float64Dtype, pd.Float32Dtype)):
+      |            df[_c] = df[_c].astype("float64")
+      |    return df""".stripMargin
+  )
+
   /**
     * What a source writes where the file it reads should be named.
     *