From c83d9c763fe356838babcaabb705e36ba32592ed Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Mon, 5 Oct 2026 14:13:38 +0200 Subject: [PATCH] Carry a bytea on the typed SQL surfaces The typed Spark and Flink SQL surfaces of tools/codegen_jvm.py carry the SQL type bytea as a Java byte[], BinaryType in Spark and BYTES in Flink. A byte array a function reads, shape.inputArrays over uint8_t with its length in another parameter, is the bytea the SQL argument holds, which the jar copies with its length (span_from_wkb(byte[])); a byte array a function returns, its length in a size out-parameter, is the byte[] the jar copies out of the buffer MEOS allocated and frees. The FromBinary and FromEWKB readers, the _send writers, the raquet reader over bytes and the raster tile value over pixels are registered. Witness: over the catalog MEOS-API 18a1e6d53c derives from MobilityDB master e464861d66, the surfaces generated by main register no reader from a bytea and no writer to one, refusing tfloatFromBinary(bytea) and its siblings as "arity:jmeos", the jar's one byte[] against the two C parameters, and temporal_send(tfloat) and its siblings as "ret:bytea/uint8_t *". With this change the typed Spark surface answers asText(tfloatFromBinary(temporal_send(tfloatFromText( '[1@2001-01-01, 2@2001-01-02]')))) with the value read, the same for a tgeompoint written as EWKB with its SRID, an intspan and an stbox, and tfloatFromBinary(X'00') raises "Invalid temporal type code in WKB string: 0". Measured over that catalog and the jar it builds: both surfaces register 1128 SQL functions and 7852 overloads, from 999 and 7704, the 129 functions added being the binary readers and writers and pixels and rasterTileValueQuadbin, and raquetRead gains its form over bytes; no other generated file differs. No signature is refused as "ret:bytea" any longer, from 55, and "arity:jmeos" falls to 30, from 183; the signatures it no longer stops reach the checks after it, "sqltype:internal" (the binary receivers over internal) and "arity:sql". Why: a bytea is how SQL hands bytes, and the jar already passes and returns them as byte[], so a binding carries one as the other. --- tools/codegen_jvm.py | 41 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/tools/codegen_jvm.py b/tools/codegen_jvm.py index 8841cb6ae..0cabcc3c9 100644 --- a/tools/codegen_jvm.py +++ b/tools/codegen_jvm.py @@ -541,7 +541,7 @@ def run_facades(args): 'boolean': 'Boolean', 'integer': 'Integer', 'smallint': 'Short', 'bigint': 'Long', 'float': 'Double', 'double precision': 'Double', 'text': 'String', 'cstring': 'String', 'timestamptz': 'java.time.Instant', 'date': 'java.time.LocalDate', - 'interval': 'java.time.Duration', + 'interval': 'java.time.Duration', 'bytea': 'byte[]', } # The Flink SQL type of each Flink-side scalar class; a MEOS value class carries its own TYPE. @@ -551,6 +551,7 @@ def run_facades(args): 'Double': 'DataTypes.DOUBLE()', 'String': 'DataTypes.STRING()', 'java.time.Instant': 'DataTypes.TIMESTAMP_LTZ(6)', 'java.time.LocalDate': 'DataTypes.DATE()', 'java.time.Duration': 'DataTypes.INTERVAL(DataTypes.SECOND(3))', + 'byte[]': 'DataTypes.BYTES()', } # The WKB variant the WKB writers take: the extended form, which keeps the SRID. @@ -570,6 +571,10 @@ def run_facades(args): def _datatype(cls): + # A class the table names (byte[], a bytea) before an array of its element, as + # #_spark_datatype reads SPARK_DATATYPE first. + if cls in SQL_DATATYPE: + return SQL_DATATYPE[cls] if cls.endswith('[]'): return f'DataTypes.ARRAY({_datatype(cls[:-2])})' return SQL_DATATYPE.get(cls, f'{cls}.TYPE') @@ -664,8 +669,13 @@ def row_of(self): else 'org.apache.flink.types.Row.of') def visible(self, f): + """(the C parameters a SQL argument or a bound literal stands for, the out-parameters). + The length of a byte array (shape.inputArrays over uint8_t) is neither: the jar folds + the array and its length into one byte[] (span_from_wkb(byte[])), which a SQL bytea + stands for, as #_inputs passes the count of any other input array itself.""" outs = set((f.get('shape') or {}).get('outParams') or []) - return [p for p in f['params'] if p['name'] not in outs], \ + folded = set(_byte_arrays(f).values()) + return [p for p in f['params'] if p['name'] not in outs and p['name'] not in folded], \ [p for p in f['params'] if p['name'] in outs] def signatures(self, f): @@ -830,6 +840,9 @@ def _arg(m, sql, p, jt, name, temps): return t if sql == 'boolean' and jt == 'boolean': return name + if sql == 'bytea' and jt == 'byte[]' and b == 'uint8_t': + # The WKB a reader takes, which the jar copies with its length (#_byte_arrays). + return name if sql in ('integer', 'smallint') and jt in ('int', 'short', 'byte'): return name if jt == 'int' else f'({jt}) (int) {name}' if sql in ('integer', 'bigint') and jt == 'long': @@ -930,6 +943,10 @@ def _ret(m, sql, f, jt, outs): ['if (_r == null) return null;', 'try { return GeneratedFunctions.text_out(_r); }', 'finally { MeosSqlRuntime.freeResult(_r, _in); }']) + if sql == 'bytea' and jt == 'byte[]': + # The WKB a writer returns, which the jar copies out of the buffer MEOS allocated and + # frees, its length read from the size out-parameter. + return 'byte[]', ['return _r;'] if sql == 'interval' and jt == 'jnr.ffi.Pointer': return ('java.time.Duration', ['if (_r == null) return null;', @@ -966,6 +983,16 @@ def _flink_class(m, sql): return SQL_SCALAR.get(sql) +def _byte_arrays(f): + """{byte array parameter: its length parameter} over the shape.inputArrays of `f` whose + element is uint8_t, the WKB a reader takes (span_from_wkb(wkb, size)), which the jar folds + into one byte[]; every other input array is passed by #_array_arg.""" + return {a['param']: (a.get('lengthFrom') or {}).get('name') + for a in (f.get('shape') or {}).get('inputArrays') or [] + if _base((a.get('element') or {}).get('canonical')) == 'uint8_t' + and (a.get('lengthFrom') or {}).get('kind') == 'param'} + + def _array_arg(m, elem, a, p, jt, name, temps): """(Flink class, Java expression) passing the SQL array `name` of `elem` to the jar parameter shape.inputArrays pairs it with (entry `a`), or None. @@ -1053,7 +1080,9 @@ def _inputs(m, f, args, jtypes, bound=None, order=None): to 0).""" vis, _ = m.visible(f) jsig = {'arg_types': jtypes} - arrays = {a['param']: a for a in (f.get('shape') or {}).get('inputArrays') or []} + folded = _byte_arrays(f) + arrays = {a['param']: a for a in (f.get('shape') or {}).get('inputArrays') or [] + if a['param'] not in folded} counts = {} for a in arrays.values(): lf = a.get('lengthFrom') or {} @@ -1904,12 +1933,14 @@ def run_flink_sql(args): 'Short': 'DataTypes.ShortType', 'Long': 'DataTypes.LongType', 'Double': 'DataTypes.DoubleType', 'String': 'DataTypes.StringType', 'java.time.Instant': 'DataTypes.TimestampType', 'java.time.LocalDate': 'DataTypes.DateType', - 'java.time.Duration': 'MeosSqlRuntime.DURATION', + 'java.time.Duration': 'MeosSqlRuntime.DURATION', 'byte[]': 'DataTypes.BinaryType', } def _spark_datatype(m, cls): """The Spark DataType expression of a Java class #_overload names, or None.""" + if cls in SPARK_DATATYPE: + return SPARK_DATATYPE[cls] if cls.endswith('[]'): inner = _spark_datatype(m, cls[:-2]) return inner and f'DataTypes.createArrayType({inner})' @@ -1929,6 +1960,8 @@ def _spark_arg(cls, i): return f'MeosSqlRuntime.instants({a})' if cls == 'java.time.LocalDate[]': return f'MeosSqlRuntime.localDates({a})' + if cls == 'byte[]': + return f'(byte[]) {a}' if cls.endswith('[]'): return f'MeosSqlRuntime.array({a}, new {cls[:-2]}[0])' return f'({cls}) {a}'