Running a Query on Apache Spark and Apache Flink

Written with functions only, a query no longer depends on operators. Spark and Flink differ from PostgreSQL in three more ways, and each asks for a change. They already define some of the names MobilityDB uses: both have round, lower and length, and Flink refuses set and unnest without quotes; such a name takes a prefix naming the class or base type of its argument (the section called “Names the Engines Own”). They never call the input or output function of a type for you, so a literal, a cast and the text of a value become function calls the query writes (the section called “Reading and Writing Values”). And each engine calls a function returning rows in its own way (the section called “Functions Returning Rows”). The following table sums up these changes; a function it does not list keeps its name.

What the query holdsMobilityDBSpark and Flink
An operator: its named function, as the sections above give ittrip && boxstboxOverlaps(trip, box)
A name the engine owns, over a value whose base type has the operation: the base type in frontround(speed(trip), 2)floatRound(speed(trip), 2)
A name the engine owns, over another set, span, span set or temporal value: the class in frontlower(timeSpan(trip))spanLower(timeSpan(trip))
A name the engine owns, over a box or a base type: the type in frontround(box, 2)stboxRound(box, 2)
A typed literal: the text constructor of its typetint '1@2001-01-01'tintFromText('1@2001-01-01')
The text of a value, which PostgreSQL writes through the output function of its type: asTextSELECT speed(trip)SELECT asText(speed(trip))
A cast: the function its entry gives beside it, as tbool::tint is tint(tbool)flag::tinttint(flag)

Names the Engines Own

A name the engine already defines takes the prefix of the class or base type of its argument, and functions that belong together take the same prefix: spanLowerInc goes with spanLower, geoCumulativeLength with geoLength. Spark and Flink use the same names, so a query written for one runs on the other, and round(1.23456, 2) keeps meaning the engine's own rounding of a number. The names that change are the following.

MobilityDB nameClass of the argumentSpark and Flink name
lower, upper, initcaptextset, ttexttextLower, textUpper, textInitcap
lower, upper, lowerInc, upperIncspan, span setspanLower, spanUpper, spanLowerInc, spanUpperInc, spansetLower, spansetUpper, spansetLowerInc, spansetUpperInc
lowerInc, upperInctemporaltemporalLowerInc, temporalUpperInc
roundfloat and its set, span, span set and temporal typesfloatRound
 geometry and geography, their sets and temporal types, trgeometrygeoRound
 cbuffer, cbufferset, tcbuffer; npoint, npointset, tnpoint; nsegment; pose, poseset, tpose; posechain, posechainset, tposechain; tbox; stbox; tpcboxcbufferRound, npointRound, nsegmentRound, poseRound, posechainRound, tboxRound, stboxRound, tpcboxRound
abstint, tbigint, tfloatintAbs, bigintAbs, floatAbs
ceil, floor, degrees, radians, cos, sin, tan, exp, ln, log10float and its set, span, span set and temporal typesfloatCeil, floatFloor, floatDegrees, floatRadians, floatCos, floatSin, floatTan, floatExp, floatLn, floatLog10
transformthe sets and temporal types of geometry and geography, trgeometrygeoTransform
transformPipelinegeometry and geography, their sets and temporal types, trgeometrygeoTransformPipeline
 cbuffer, cbufferset, tcbuffer; pose, poseset, tpose; posechain, posechainset, tposechain; stbox; rastercbufferTransform, poseTransform, posechainTransform, stboxTransform, rasterTransform, and the …TransformPipeline of each
translate, affine, rotate, rotateX, rotateY, rotateZ, scale, transscalegeometry, tgeompoint, tgeometrygeoTranslate, geoAffine, geoRotate, geoRotateX, geoRotateY, geoRotateZ, geoScale, geoTransscale
translate, rotate, rotateX, rotateY, rotateZtcbuffer; tpose; tposechaincbufferTranslate, cbufferRotate, cbufferRotateZ; poseTranslate, poseRotate, poseRotateX, poseRotateY, poseRotateZ; the same for posechain (posechainTranslate, …)
 trgeometrygeoTranslate, geoRotate, geoRotateX, geoRotateY, geoRotateZ
width, height, bandPixelType, bandHasNoDataValue, bandNoDataValue, pixels, asHexWKB, stboxraster; raquetrasterWidth, rasterHeight, rasterBandPixelType, rasterBandHasNoDataValue, rasterBandNoDataValue, rasterPixels, rasterAsHexWKB, rasterStbox; raquetWidth, raquetHeight, raquetBandPixelType, raquetBandHasNoDataValue, raquetBandNoDataValue, raquetPixels, raquetAsHexWKB, raquetStbox
length, cumulativeLengthgeometry, geography, tgeompoint, tgeogpoint, trgeometrygeoLength, geoCumulativeLength
 tnpointnpointLength, npointCumulativeLength
contains, covers, disjoint, intersects, touches, dwithin, relategeometry, geographygeoContains, geoCovers, geoDisjoint, geoIntersects, geoTouches, geoDwithin, geoRelate
reversegeometrygeoReverse
collectgeometry[]geoCollect
transformgeometry, geographygeoTransform
hash, hashExtendedset, span, span set, box, temporal, base typesetHash, spanHash, spansetHash, tboxHash, stboxHash, temporalHash, cbufferHash, …, and their …HashExtended
insert, update, mergetemporaltemporalInsert, temporalUpdate, temporalMerge
tMin, tMax, merge, appendInstant, appendSequence, tAnd, tOr (the aggregates)temporaltMinAgg, tMaxAgg, mergeAgg, appendInstantAgg, appendSequenceAgg, tAndAgg, tOrAgg
setUnion, spanUnion, spansetUnion (the aggregates)base value, set, span, span setsetUnionAgg, spanUnionAgg, spansetUnionAgg
sameAlignment (the aggregate)rastersameAlignmentAgg
wAvg (the aggregate)tint, tbigint, tfloatwAvgAgg
unnestset, temporalsetUnnest, temporalUnnest
set (the constructor)anysetMake

The names a raster and a raquet share take the prefix of their type too, even where the engine defines no such name: both travel as WKB, so an engine holding one function per name cannot tell a raster given to width from a raquet.

Reading and Writing Values

PostgreSQL reads and writes a value through the four functions its type declares, and calls them for you: a typed literal, a cast from text and a text COPY go through the input function, the text a client displays through the output function, and a binary COPY or a driver in binary mode through the receive and send functions. Spark and Flink call none of them. A value travels between them as its Well-Known Binary (WKB) bytes, and each conversion is a function the query calls, each with its inverse, as the following table shows.

What PostgreSQL usesDirectionSpark and Flink
the input function: a typed literal, a cast from texttext to valuettypeFromText(text), settypeFromText(text), boxFromText(text), …, one per type, prefixed by its name; with an SRID also tspatialFromEWKT(text)
the output function: the text a client printsvalue to textasText(value); with an SRID also asEWKT(value)
the receive function: binary COPY, a binary driverbinary to valuettypeFromBinary(bytea), ttypeFromHexWKB(text), and likewise for every other type; with an SRID also tspatialFromEWKB(bytea), tspatialFromHexEWKB(text)
the send functionvalue to binaryasBinary(value), asHexWKB(value); with an SRID also asEWKB(value), asHexEWKB(value)
noneMF-JSONttypeFromMFJSON(text), asMFJSON(ttype)

Every type has the functions of the table that apply to it: the E forms when it carries an SRID its plain text and binary forms omit, and MF-JSON when it is temporal. This includes the types PostgreSQL reads and writes only through its four functions: the set, span and span set types of integers, big integers, floats, dates and timestamps, textset and jsonbset, tbox and stbox, tbool, tint, tbigint, tfloat and ttext, and quadbin, s2cell, nsegment and tpcbox.

The text of a raquet is its hexadecimal WKB, so it is read with raquetFromHexWKB and written with asHexWKB, as a PostGIS raster is with ST_RastFromHexWKB and ST_AsHexWKB. A box, stbox or tpcbox, writes its SRID in both its text and its WKB, and so has no E form. Every WKB output takes the byte order endian, and every text output of a value with floats or coordinates the number of decimal digits maxdecimaldigits. A column of WKB bytes, such as a TemporalParquet column, is read in every engine with the binary constructor of its type, as in tgeompointFromBinary(col).

Functions Returning Rows

A function returning several rows, such as timeSplit or unnest, is called in PostgreSQL as a table in the FROM clause. In Spark and Flink the same function, unnest under the names setUnnest and temporalUnnest, returns an array with one element per row, an array of records when it returns several columns. Flink turns the array into rows with CROSS JOIN UNNEST, and Spark with LATERAL VIEW explode, or inline for an array of records.

Example

The query

SELECT vehicle, round(length(trip)::numeric, 1), lower(timeSpan(trip))
FROM Trips
WHERE trip && stbox 'SRID=3857;STBOX X((0,0),(1000,1000))';

is written, in Spark and Flink,

SELECT vehicle, round(geoLength(trip), 1), spanLower(timeSpan(trip))
FROM Trips
WHERE stboxOverlaps(trip, stboxFromText('SRID=3857;STBOX X((0,0),(1000,1000))'));

Here round applies to a number, so it is the engine's own function and keeps its name.